utils/validate-admin-api.py
Rama master · réplica de lectura del repositorio original.
#!/usr/bin/env python3
# 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.
"""Validate the checked-in Admin API contract; never download contract inputs."""
import argparse
from copy import deepcopy
import json
from pathlib import Path
import re
import shlex
import sys
from urllib.parse import unquote, urlsplit
from jsonschema import Draft202012Validator
from jsonschema.exceptions import SchemaError
from openapi_spec_validator import OpenAPIV31SpecValidator
from referencing import Registry, Resource
from referencing.jsonschema import DRAFT202012
ROOT = Path(__file__).resolve().parents[1]
METHODS = {'get', 'put', 'post', 'delete', 'patch', 'head', 'options', 'trace'}
BASE = 'urn:apisix:admin-api'
# These contain instance data, not contract objects or schemas.
DATA = {'example', 'default', 'enum', 'const', 'value'}
MAPS = {'properties', 'patternProperties', '$defs', 'dependentSchemas', 'schemas',
'paths', 'responses', 'content', 'headers', 'links', 'callbacks', 'pathItems',
'requestBodies', 'parameters', 'securitySchemes', 'examples'}
def objects(value, path='#', mapping=False):
"""Walk contract objects, keeping arbitrary names and instance data separate."""
if isinstance(value, dict):
if not mapping:
yield path, value
for key, child in value.items():
if mapping or (key not in DATA and not key.startswith('x-')):
if not mapping and key == 'examples' and isinstance(child, list):
continue # JSON Schema examples are instance data.
yield from objects(child, path + '/' + key.replace('~', '~0').replace('/', '~1'),
not mapping and key in MAPS and isinstance(child, dict))
elif isinstance(value, list):
for i, child in enumerate(value):
yield from objects(child, path + '/' + str(i))
def resolve(document, ref):
if not isinstance(ref, str) or not ref.startswith('#/'):
raise ValueError(f'only internal JSON Pointer refs are allowed: {ref!r}')
result = document
for token in unquote(ref[2:]).split('/'):
token = token.replace('~1', '/').replace('~0', '~')
try:
result = result[int(token)] if isinstance(result, list) else result[token]
except (KeyError, IndexError, ValueError, TypeError) as exc:
raise ValueError(f'unresolved ref: {ref}') from exc
return result
def dereference(document, obj):
seen = set()
while isinstance(obj, dict) and '$ref' in obj:
ref = obj['$ref']
if ref in seen:
raise ValueError(f'cyclic Reference Object: {ref}')
seen.add(ref)
obj = resolve(document, ref)
return obj
def absolute_refs(value):
result = deepcopy(value)
for _, obj in objects(result):
for key in ('$ref', '$dynamicRef'):
if key in obj:
obj[key] = BASE + obj[key]
return result
def validate(document, versions):
"""Return actionable errors. Local ref preflight runs before the OAS library."""
errors = []
if document.get('openapi') != '3.1.0':
errors.append('openapi must be 3.1.0')
version = document.get('info', {}).get('version')
for source, expected in versions.items():
if version != expected:
errors.append(f'info.version {version!r} differs from {source}: {expected!r}')
for path, obj in objects(document):
for key in ('$ref', '$dynamicRef'):
if key in obj:
try:
resolve(document, obj[key])
except ValueError as exc:
errors.append(f'{path}: {exc}')
if '$id' in obj:
errors.append(f'{path}: embedded $id changes local ref scope; use document pointers')
if errors:
return errors
schemas = list(document.get('components', {}).get('schemas', {}).items())
schemas += [(path + '/schema', obj['schema']) for path, obj in objects(document)
if 'schema' in obj]
for path, schema in schemas:
try:
Draft202012Validator.check_schema(schema)
except SchemaError as exc:
errors.append(f'{path}: invalid JSON Schema: {exc.message}')
if errors:
return errors
errors.extend('OpenAPI: ' + str(e) for e in OpenAPIV31SpecValidator(document).iter_errors())
if errors:
return errors
registry = Registry().with_resource(
BASE, Resource.from_contents(absolute_refs(document), default_specification=DRAFT202012))
def check_value(schema, value, location):
validator = Draft202012Validator(absolute_refs(schema), registry=registry)
errors.extend(f'{location}: {e.message}' for e in validator.iter_errors(value))
def check_schema(schema, location):
for path, obj in objects(schema, location):
if 'default' in obj:
check_value(obj, obj['default'], path + '/default')
if 'examples' in obj:
for i, example in enumerate(obj['examples']):
check_value(obj, example, path + f'/examples/{i}')
if 'example' in obj:
check_value(obj, obj['example'], path + '/example')
# Components plus schemas on parameters, request bodies, response media and headers.
for name, schema in document.get('components', {}).get('schemas', {}).items():
check_schema(schema, '#/components/schemas/' + name)
for path, obj in objects(document):
if 'schema' not in obj:
continue
schema = obj['schema']
check_schema(schema, path + '/schema')
if 'example' in obj:
check_value(schema, obj['example'], path + '/example')
for name, example in obj.get('examples', {}).items():
example = dereference(document, example)
if 'externalValue' in example:
errors.append(f'{path}/examples/{name}: use a checked-in value, not externalValue')
elif 'value' in example:
check_value(schema, example['value'], path + '/examples/' + name)
ids = set()
for path, raw_item in document.get('paths', {}).items():
item = dereference(document, raw_item)
for method in METHODS & item.keys():
op = item[method]
location = f'{method.upper()} {path}'
if not op.get('responses'):
errors.append(f'{location}: responses must be non-empty')
op_id = op.get('operationId')
if not isinstance(op_id, str) or not re.fullmatch(r'[A-Za-z][A-Za-z0-9]*', op_id) or op_id in ids:
errors.append(f'{location}: missing or duplicate operationId {op_id!r}')
ids.add(op_id)
params = {}
for group in (item.get('parameters', []), op.get('parameters', [])):
for param in group:
param = dereference(document, param)
params[(param['in'], param['name'])] = param
declared = {name for (where, name) in params if where == 'path'}
if declared != set(re.findall(r'{([^{}]+)}', path)):
errors.append(f'{location}: path parameters do not match template')
for (where, name), param in params.items():
if where == 'path' and param.get('required') is not True:
errors.append(f'{location}: path parameter {name} must be required')
for field in ('requestBody', 'responses'):
entries = op.get('responses', {}).values() if field == 'responses' else [op.get(field)]
for entry in entries:
if entry is None:
continue
entry = dereference(document, entry)
for media, content in entry.get('content', {}).items():
if 'schema' not in content:
errors.append(f'{location} {field} {media}: missing schema')
for sample in op.get('x-codeSamples', []):
if sample.get('lang', '').lower() != 'curl':
continue
try:
args = shlex.split(sample['source'])
url = next(x for x in args if x.startswith(('http://', 'https://')))
expected_path = re.sub(r'{[^}]+}', r'[^/]+', path)
if not re.fullmatch(expected_path, urlsplit(url).path):
errors.append(f'{location}: cURL URL does not match operation')
actual_method = args[args.index('-X') + 1].lower() if '-X' in args else 'get'
if actual_method != method:
errors.append(f'{location}: cURL method does not match operation')
for flag in ('-d', '--data', '--data-raw'):
if flag in args:
body = json.loads(args[args.index(flag) + 1])
schema = dereference(document, op['requestBody'])['content']['application/json']['schema']
check_value(schema, body, location + '/curl')
except (ValueError, KeyError, StopIteration) as exc:
errors.append(f'{location}: invalid cURL sample: {exc}')
return errors
def repository_versions(root):
versions = {f'docs/{lang}/latest/config.json': json.loads(
(root / f'docs/{lang}/latest/config.json').read_text())['version'] for lang in ('en', 'zh')}
text = (root / 'apisix/core/version.lua').read_text()
versions['apisix/core/version.lua'] = re.search(r'VERSION\s*=\s*"([^"]+)"', text).group(1)
return versions
def validate_source(document, root):
"""Compare structural constraints with this checkout's pure Lua schema table.
Lupa evaluates schema_def.lua only: no APISIX module or network is loaded.
This is not an OpenResty integration test or a replacement for Lua checkers.
"""
from lupa import LuaRuntime, lua_type
def from_lua(value):
if lua_type(value) != 'table':
return value
keys = list(value.keys())
if keys and all(isinstance(key, int) for key in keys):
return [from_lua(value[i]) for i in range(1, len(keys) + 1)]
result = {key: from_lua(value[key]) for key in keys}
if 'dependencies' in result:
dependencies = result.pop('dependencies')
for keyword, typ in [('dependentSchemas', dict), ('dependentRequired', list)]:
entries = {k: v for k, v in dependencies.items() if isinstance(v, typ)}
if entries:
result[keyword] = entries
return result
lua = LuaRuntime()
lua_schema = lua.execute((root / 'apisix/schema_def.lua').read_text())
source = from_lua(lua_schema)
def constraints(value):
if isinstance(value, list):
return [constraints(item) for item in value]
if not isinstance(value, dict):
return value
if '$ref' in value:
return constraints(resolve(document, value['$ref']))
result = {k: constraints(v) for k, v in value.items()
if k not in {'description', 'title', 'readOnly', 'examples', 'example'}
and not k.startswith('x-')}
# APISIX's JSON Schema validator skips structural validation for
# $secret:// and $env:// values. OpenAPI models that runtime hook
# explicitly, either as an accepted union branch or as an exclusion
# where later runtime validation requires an inline PEM value.
secret_reference = {
'type': 'string',
'pattern': r'^\$(?:secret|[eE][nN][vV])://',
}
if result.get('not') == secret_reference:
result.pop('not')
if set(result) == {'anyOf'} and len(result['anyOf']) == 2:
branches = [branch for branch in result['anyOf']
if branch != secret_reference]
if len(branches) == 1:
return branches[0]
return result
names = {'ResourceId': 'id_schema', 'Route': 'route', 'Service': 'service',
'Upstream': 'upstream', 'Consumer': 'consumer', 'Credential': 'credential',
'ConsumerGroup': 'consumer_group', 'GlobalRule': 'global_rule',
'PluginConfig': 'plugin_config', 'SSL': 'ssl', 'StreamRoute': 'stream_route',
'Proto': 'proto', 'GraphQLCostDecoration': 'graphql_cost_decoration'}
errors = []
schemas = document['components']['schemas']
for name, key in names.items():
expected = constraints(source[key])
# resource.check_conf injects the path id before running these schemas.
if name in {'ConsumerGroup', 'GlobalRule', 'PluginConfig'}:
expected['required'].remove('id')
if constraints(schemas[name]) != expected:
errors.append(f'{name}: structural constraints differ from schema_def.lua ({key})')
lua.globals().core = lua.table_from({'schema': lua_schema})
for name, backend in [('VaultSecret', 'vault'), ('AwsSecret', 'aws'), ('GcpSecret', 'gcp')]:
text = (root / f'apisix/secret/{backend}.lua').read_text()
declaration = 'local schema =' + text.split('local schema =', 1)[1].split('local _M', 1)[0]
expected = from_lua(lua.execute(declaration + '\nreturn schema'))
if constraints(schemas[name]) != constraints(expected):
errors.append(f'{name}: structural constraints differ from secret/{backend}.lua')
# Check supported canonical CRUD methods against resource.new declarations.
modules = {'routes': 'routes', 'services': 'services', 'upstreams': 'upstreams',
'consumers': 'consumers', 'ssls': 'ssl', 'global_rules': 'global_rules',
'consumer_groups': 'consumer_group', 'plugin_configs': 'plugin_config',
'plugin_metadata': 'plugin_metadata', 'stream_routes': 'stream_routes',
'protos': 'proto', 'secrets': 'secrets', 'credentials': 'credentials'}
for resource, module in modules.items():
text = (root / f'apisix/admin/{module}.lua').read_text()
match = re.search(r'unsupported_methods\s*=\s*{([^}]+)}', text)
unsupported = set(re.findall(r'"([a-z]+)"', match.group(1))) if match else set()
actual = set()
for path, item in document['paths'].items():
if resource == 'credentials':
relevant = '/consumers/{username}/credentials' in path
else:
relevant = path.startswith('/apisix/admin/' + resource) and (
path == '/apisix/admin/' + resource or
path.startswith('/apisix/admin/' + resource + '/'))
if resource == 'consumers' and '/credentials' in path:
relevant = False
if relevant:
actual.update(METHODS & item.keys())
expected = {'get', 'put', 'post', 'patch', 'delete'} - unsupported
if actual != expected:
errors.append(f'{resource}: documented methods {sorted(actual)} != {sorted(expected)}')
# Resource.put accepts a body ID when the canonical item URL is omitted.
expected_ops = set()
for resource, module in modules.items():
if resource == 'credentials':
base = '/apisix/admin/consumers/{username}/credentials'
else:
base = '/apisix/admin/' + resource
text = (root / f'apisix/admin/{module}.lua').read_text()
match = re.search(r'unsupported_methods\s*=\s*{([^}]+)}', text)
unsupported = set(re.findall(r'"(\w+)"', match.group(1))) if match else set()
suffix = '/{username}' if resource == 'consumers' else '/{plugin_name}' if resource == 'plugin_metadata' else '/{secret_type}/{id}' if resource == 'secrets' else '/{id}'
item = base + suffix
expected_ops.update([(base, 'get'), (item, 'get'), (item, 'put'), (item, 'delete')])
if resource not in ('secrets', 'plugin_metadata'):
expected_ops.add((base, 'put'))
if 'post' not in unsupported:
expected_ops.add((base, 'post'))
if 'patch' not in unsupported:
expected_ops.update([(item, 'patch'), (item + '/{sub_path}', 'patch')])
expected_ops.add(('/apisix/admin/secrets/{secret_type}', 'get'))
expected_ops.update({
('/apisix/admin/plugins/{plugin_name}', 'get'),
('/apisix/admin/schema/plugins/{plugin_name}', 'get'),
('/apisix/admin/plugins', 'get'), ('/apisix/admin/plugins/list', 'get'),
('/apisix/admin/plugins/reload', 'put'),
('/apisix/admin/schema/{resource}', 'get'),
('/apisix/admin/schema/validate/{resource}', 'post'),
('/apisix/admin', 'head'), ('/apisix/admin/configs', 'get'),
('/apisix/admin/configs', 'put'), ('/apisix/admin/configs', 'head'),
('/apisix/admin/configs/validate', 'post'),
('/apisix/admin/services/{service_id}/graphql_cost_decorations', 'get'),
('/apisix/admin/services/{service_id}/graphql_cost_decorations', 'post'),
('/apisix/admin/services/{service_id}/graphql_cost_decorations/{decoration_id}', 'get'),
('/apisix/admin/services/{service_id}/graphql_cost_decorations/{decoration_id}', 'put'),
('/apisix/admin/services/{service_id}/graphql_cost_decorations/{decoration_id}', 'patch'),
('/apisix/admin/services/{service_id}/graphql_cost_decorations/{decoration_id}', 'delete'),
})
actual_ops = {(path, method) for path, item in document['paths'].items()
for method in METHODS & item.keys()}
if actual_ops != expected_ops:
errors.append(f'operation surface differs: missing {sorted(expected_ops - actual_ops)}, extra {sorted(actual_ops - expected_ops)}')
return errors
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('spec', nargs='?', type=Path, default=ROOT / 'docs/en/latest/admin-api.json')
args = parser.parse_args()
spec_text = args.spec.read_text()
if re.search(r'api7|apiseven', spec_text, re.IGNORECASE):
print('Admin API specification must not contain API7-specific references', file=sys.stderr)
return 1
document = json.loads(spec_text)
errors = validate(document, repository_versions(ROOT))
if not errors:
errors.extend(validate_source(document, ROOT))
if errors:
print('\n'.join(errors), file=sys.stderr)
return 1
print('Admin API OpenAPI 3.1 contract is valid (refs, operations, schemas, examples, defaults, versions, release source constraints and methods).')
return 0
if __name__ == '__main__':
sys.exit(main())