diff options
Diffstat (limited to 'tests/api/fuzzer.py')
| -rw-r--r-- | tests/api/fuzzer.py | 378 |
1 files changed, 378 insertions, 0 deletions
diff --git a/tests/api/fuzzer.py b/tests/api/fuzzer.py new file mode 100644 index 0000000..ee12a02 --- /dev/null +++ b/tests/api/fuzzer.py @@ -0,0 +1,378 @@ +import argparse +import json +import logging +import posixpath +import random +import re +import requests +import string +import sys +import urllib.parse + +####################################################################################################################### +# Utilities + + +def some(s): + return random.choice(sorted(s)) + + +def not_some(s): + test_set = random.choice([string.ascii_uppercase + string.ascii_lowercase, + string.digits, + string.digits + ".E-", + '0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJK' + 'LMNOPQRSTUVWXYZ!"#$%\'()*+,-./:;<=>?@[\\]^_`{|}~ ']) + test_len = random.choice([1, 2, 3, 37, 61, 121]) + while True: + x = ''.join([random.choice(test_set) for _ in range(test_len)]) + if x not in s: + return x + + +def build_url(host_maybe_scheme, base_path): + try: + if '//' not in host_maybe_scheme: + host_maybe_scheme = '//' + host_maybe_scheme + url_tuple = urllib.parse.urlparse(host_maybe_scheme) + if base_path[0] == '/': + base_path = base_path[1:] + return url_tuple.netloc, posixpath.join(url_tuple.path, base_path) + except Exception as e: + L.error(f"Critical failure decoding arguments -> {e}") + sys.exit(-1) + + +####################################################################################################################### +# Data-model and processing + + +class Param(object): + def __init__(self, name, location, kind): + self.location = location + self.kind = kind + self.name = name + self.values = set() + + def dump(self): + print(f"{self.name} in {self.location} is {self.kind} : {{{self.values}}}") + + +def does_response_fit_schema(schema_path, schema, resp): + '''The schema_path argument tells us where we are (globally) in the schema. The schema argument is the + sub-tree within the schema json that we are validating against. The resp is the json subtree from the + target host's response. + + The basic idea is this: swagger defines a model of valid json trees. In this sense it is a formal + language and we can validate a given server response by checking if the language accepts a particular + server response. This is basically a parser, but instead of strings we are operating on languages + of trees. + + This could probably be extended to arbitrary swagger definitions - but the amount of work increases + rapidly as we attempt to cover the full semantics of languages of trees defined in swagger. Instead + we have some special cases that describe the parts of the semantics that we've used to describe the + netdata API. + + If we hit an error (in the schema) that prevents further checks then we return early, otherwise we + try to collect as many errors as possible. + ''' + success = True + if "type" not in schema: + L.error(f"Cannot progress past {schema_path} -> no type specified in dictionary") + print(json.dumps(schema, indent=2)) + return False + if schema["type"] == "object": + if isinstance(resp, dict) and "properties" in schema and isinstance(schema["properties"], dict): + L.debug(f"Validate properties against dictionary at {schema_path}") + for k, v in schema["properties"].items(): + L.debug(f"Validate {k} received with {v}") + if v.get("required", False) and k not in resp: + L.error(f"Missing {k} in response at {schema_path}") + print(json.dumps(resp, indent=2)) + return False + if k in resp: + if not does_response_fit_schema(posixpath.join(schema_path, k), v, resp[k]): + success = False + elif isinstance(resp, dict) and "additionalProperties" in schema \ + and isinstance(schema["additionalProperties"], dict): + kv_schema = schema["additionalProperties"] + L.debug(f"Validate additionalProperties against every value in dictionary at {schema_path}") + if "type" in kv_schema and kv_schema["type"] == "object": + for k, v in resp.items(): + if not does_response_fit_schema(posixpath.join(schema_path, k), kv_schema, v): + success = False + else: + L.error("Don't understand what the additionalProperties means (it has no type?)") + return False + else: + L.error(f"Can't understand schema at {schema_path}") + print(json.dumps(schema, indent=2)) + return False + elif schema["type"] == "string": + if isinstance(resp, str): + L.debug(f"{repr(resp)} matches {repr(schema)} at {schema_path}") + return True + L.error(f"{repr(resp)} does not match schema {repr(schema)} at {schema_path}") + return False + elif schema["type"] == "boolean": + if isinstance(resp, bool): + L.debug(f"{repr(resp)} matches {repr(schema)} at {schema_path}") + return True + L.error(f"{repr(resp)} does not match schema {repr(schema)} at {schema_path}") + return False + elif schema["type"] == "number": + if 'nullable' in schema and resp is None: + L.debug(f"{repr(resp)} matches {repr(schema)} at {schema_path} (because nullable)") + return True + if isinstance(resp, int) or isinstance(resp, float): + L.debug(f"{repr(resp)} matches {repr(schema)} at {schema_path}") + return True + L.error(f"{repr(resp)} does not match schema {repr(schema)} at {schema_path}") + return False + elif schema["type"] == "integer": + if 'nullable' in schema and resp is None: + L.debug(f"{repr(resp)} matches {repr(schema)} at {schema_path} (because nullable)") + return True + if isinstance(resp, int): + L.debug(f"{repr(resp)} matches {repr(schema)} at {schema_path}") + return True + L.error(f"{repr(resp)} does not match schema {repr(schema)} at {schema_path}") + return False + elif schema["type"] == "array": + if "items" not in schema: + L.error(f"Schema for array at {schema_path} does not specify items!") + return False + item_schema = schema["items"] + if not isinstance(resp, list): + L.error(f"Server did not return a list for {schema_path} (typed as array in schema)") + return False + for i, item in enumerate(resp): + if not does_response_fit_schema(posixpath.join(schema_path, str(i)), item_schema, item): + success = False + else: + L.error(f"Invalid swagger type {schema['type']} for {type(resp)} at {schema_path}") + print(json.dumps(schema, indent=2)) + return False + return success + + +class GetPath(object): + def __init__(self, url, spec): + self.url = url + self.req_params = {} + self.opt_params = {} + self.success = None + self.failures = {} + if 'parameters' in spec.keys(): + for p in spec['parameters']: + name = p['name'] + req = p.get('required', False) + target = self.req_params if req else self.opt_params + target[name] = Param(name, p['in'], p['type']) + if 'default' in p: + defs = p['default'] + if isinstance(defs, list): + for d in defs: + target[name].values.add(d) + else: + target[name].values.add(defs) + if 'enum' in p: + for v in p['enum']: + target[name].values.add(v) + if req and len(target[name].values) == 0: + print(f"FAIL: No default values in swagger for required parameter {name} in {self.url}") + for code, schema in spec['responses'].items(): + if code[0] == "2" and 'schema' in schema: + self.success = schema['schema'] + elif code[0] == "2": + L.error(f"2xx response with no schema in {self.url}") + else: + self.failures[code] = schema + + def generate_success(self, host): + url_args = "&".join([f"{p.name}={some(p.values)}" for p in self.req_params.values()]) + base_url = urllib.parse.urljoin(host, self.url) + test_url = f"{base_url}?{url_args}" + if url_filter.match(test_url): + try: + resp = requests.get(url=test_url, verify=(not args.tls_no_verify)) + self.validate(test_url, resp, True) + except Exception as e: + L.error(f"Network failure in test {e}") + else: + L.debug(f"url_filter skips {test_url}") + + def generate_failure(self, host): + all_params = list(self.req_params.values()) + list(self.opt_params.values()) + bad_param = ''.join([random.choice(string.ascii_lowercase) for _ in range(5)]) + while bad_param in all_params: + bad_param = ''.join([random.choice(string.ascii_lowercase) for _ in range(5)]) + all_params.append(Param(bad_param, "query", "string")) + url_args = "&".join([f"{p.name}={not_some(p.values)}" for p in all_params]) + base_url = urllib.parse.urljoin(host, self.url) + test_url = f"{base_url}?{url_args}" + if url_filter.match(test_url): + try: + resp = requests.get(url=test_url, verify=(not args.tls_no_verify)) + self.validate(test_url, resp, False) + except Exception as e: + L.error(f"Network failure in test {e}") + + def validate(self, test_url, resp, expect_success): + try: + resp_json = json.loads(resp.text) + except json.decoder.JSONDecodeError as e: + L.error(f"Non-json response from {test_url}") + return + success_code = resp.status_code >= 200 and resp.status_code < 300 + if success_code and expect_success: + if self.success is not None: + if does_response_fit_schema(posixpath.join(self.url, str(resp.status_code)), self.success, resp_json): + L.info(f"tested {test_url}") + else: + L.error(f"tested {test_url}") + else: + L.error(f"Missing schema {test_url}") + elif not success_code and not expect_success: + schema = self.failures.get(str(resp.status_code), None) + if schema is not None: + if does_response_fit_schema(posixpath.join(self.url, str(resp.status_code)), schema, resp_json): + L.info(f"tested {test_url}") + else: + L.error(f"tested {test_url}") + else: + L.error("Missing schema for {resp.status_code} from {test_url}") + else: + L.error(f"Received incorrect status code {resp.status_code} against {test_url}") + + +def get_the_spec(url): + if url[:7] == "file://": + with open(url[7:]) as f: + return f.read() + return requests.get(url=url).text + + +# Swagger paths look absolute but they are relative to the base. +def not_absolute(path): + return path[1:] if path[0] == '/' else path + + +def find_ref(spec, path): + if len(path) > 0 and path[0] == '#': + return find_ref(spec, path[1:]) + if len(path) == 1: + return spec[path[0]] + return find_ref(spec[path[0]], path[1:]) + + +def resolve_refs(spec, spec_root=None): + '''Find all "$ref" keys in the swagger spec and inline their target schemas. + + As with all inliners this will break if a definition recursively links to itself, but this should not + happen in swagger as embedding a structure inside itself would produce a record of infinite size.''' + if spec_root is None: + spec_root = spec + newspec = {} + for k, v in spec.items(): + if k == "$ref": + path = v.split('/') + target = find_ref(spec_root, path) + # Unfold one level of the tree and erase the $ref if possible. + if isinstance(target, dict): + for kk, vv in resolve_refs(target, spec_root).items(): + newspec[kk] = vv + else: + newspec[k] = target + elif isinstance(v, dict): + newspec[k] = resolve_refs(v, spec_root) + else: + newspec[k] = v + # This is an artifact of inline the $refs when they are inside a properties key as their children should be + # pushed up into the parent dictionary. They must be merged (union) rather than replace as we use this to + # implement polymorphism in the data-model. + if 'properties' in newspec and isinstance(newspec['properties'], dict) and \ + 'properties' in newspec['properties']: + sub = newspec['properties']['properties'] + del newspec['properties']['properties'] + if 'type' in newspec['properties']: + del newspec['properties']['type'] + for k, v in sub.items(): + newspec['properties'][k] = v + return newspec + + +####################################################################################################################### +# Initialization + +random.seed(7) # Default is reproducible sequences + +parser = argparse.ArgumentParser() +parser.add_argument('--url', type=str, + default='https://raw.githubusercontent.com/netdata/netdata/master/web/api/netdata-swagger.json', + help='The URL of the API definition in swagger. The default will pull the latest version ' + 'from the main branch.') +parser.add_argument('--host', type=str, + help='The URL of the target host to fuzz. The default will read the host from the swagger ' + 'definition.') +parser.add_argument('--reseed', action='store_true', + help="Pick a random seed for the PRNG. The default uses a constant seed for reproducibility.") +parser.add_argument('--passes', action='store_true', + help="Log information about tests that pass") +parser.add_argument('--detail', action='store_true', + help="Log information about the response/schema comparisons during each test") +parser.add_argument('--filter', type=str, + default=".*", + help="Supply a regex used to filter the testing URLs generated") +parser.add_argument('--tls-no-verify', action='store_true', + help="Disable TLS certification verification to allow connection to hosts that use" + "self-signed certificates") +parser.add_argument('--dump-inlined', action='store_true', + help='Dump the inlined swagger spec instead of fuzzing. For "reasons".') + +args = parser.parse_args() +if args.reseed: + random.seed() + +spec = json.loads(get_the_spec(args.url)) +inlined_spec = resolve_refs(spec) +if args.dump_inlined: + print(json.dumps(inlined_spec, indent=2)) + sys.exit(-1) + +logging.addLevelName(40, "FAIL") +logging.addLevelName(20, "PASS") +logging.addLevelName(10, "DETAIL") +L = logging.getLogger() +handler = logging.StreamHandler(sys.stdout) +if not args.passes and not args.detail: + L.setLevel(logging.ERROR) +elif args.passes and not args.detail: + L.setLevel(logging.INFO) +elif args.detail: + L.setLevel(logging.DEBUG) +handler.setFormatter(logging.Formatter(fmt="%(levelname)s %(message)s")) +L.addHandler(handler) + +url_filter = re.compile(args.filter) + +if spec['swagger'] != '2.0': + L.error(f"Unexpected swagger version") + sys.exit(-1) +L.info(f"Fuzzing {spec['info']['title']} / {spec['info']['version']}") + +host, base_url = build_url(args.host or spec['host'], inlined_spec['basePath']) + +L.info(f"Target host is {base_url}") +paths = [] +for name, p in inlined_spec['paths'].items(): + if 'get' in p: + name = not_absolute(name) + paths.append(GetPath(posixpath.join(base_url, name), p['get'])) + elif 'put' in p: + L.error(f"Generation of PUT methods (for {name} is unimplemented") + +for s in inlined_spec['schemes']: + for p in paths: + resp = p.generate_success(s + "://" + host) + resp = p.generate_failure(s+"://"+host) |