import ast

import io

import json

import os

import subprocess

import tomllib

import warnings

from contextlib import redirect_stdout

from pathlib import Path

from tempfile import TemporaryDirectory

from typing import Any

from hypothesis import given, settings

from hypothesis import strategies as st

from openapi_python_client import Config, ErrorLevel, MetaType, generate

from openapi_python_client.config import ConfigFile

from openapi_python_client.schema import OpenAPI

FUZZ_EXAMPLES = int(os.environ.get("OPENAPI_PYTHON_CLIENT_FUZZ_EXAMPLES", "100"))

# Canary payload: if any untrusted string breaks out of its literal/docstring in generated code,

# this appears as a real AST Name/Call node instead of text inside a string constant.

CANARY = "PWNED_CANARY"

CALL_CANARY = "EXECUTED_CANARY"

DANGEROUS_STRINGS = (

"",

"'",

'"',

"\\",

"\n",

"\r\n",

"\t",

"\x00",

'"""',

"'''",

"{value!r}",

"${value}",

"{{ value }}",

"line one\\\nline two",

"\\U0000000",

"\\N{NOT A REAL NAME}",

"# type: ignore\n" + CANARY + "()",

"\n# comment with trailing slash\\",

"from os import system",

"\N{LINE SEPARATOR}",

"\N{PILE OF POO}",

f'"{CANARY}("',

f'"{CANARY}() + "',

f"'''{CANARY}()'''",

f'"""{CANARY}()"""',

f'\\"{CANARY}()',

f"\n{CANARY}()\n",

f'"; print("{CALL_CANARY}"); _fuzz = "',

f'"""\nprint("{CALL_CANARY}")\n"""',

)

UTF8_CHARACTERS = st.characters(blacklist_categories=["Cs"])

UNTRUSTED_TEXT = st.one_of(

st.sampled_from(DANGEROUS_STRINGS),

st.text(UTF8_CHARACTERS, max_size=60),

)

NONEMPTY_UNTRUSTED_TEXT = st.one_of(

st.sampled_from(tuple(value for value in DANGEROUS_STRINGS if value)),

st.text(UTF8_CHARACTERS, min_size=1, max_size=40),

)

COMPONENT_NAME = st.one_of(

st.sampled_from(tuple(value for value in DANGEROUS_STRINGS if value and "/" not in value and "~" not in value)),

st.text(UTF8_CHARACTERS, min_size=1, max_size=30).filter(lambda value: "/" not in value and "~" not in value),

)

@st.composite

def schema_strategy(draw: st.DrawFn) -> dict[str, Any]:

description = draw(UNTRUSTED_TEXT)

kind = draw(

st.sampled_from(

("string", "integer", "number", "boolean", "array", "object", "reference", "union", "const", "free_form")

)

)

value: Any

schema: dict[str, Any] = {"description": description}

if kind == "string":

value = draw(UNTRUSTED_TEXT)

schema.update({"type": "string", "default": value, "example": value})

elif kind == "integer":

value = draw(st.integers(min_value=-(2**31), max_value=2**31 - 1))

schema.update({"type": "integer", "default": value, "example": value})

elif kind == "number":

value = draw(st.floats(allow_nan=False, allow_infinity=False, width=32))

schema.update({"type": "number", "default": value, "example": value})

elif kind == "boolean":

value = draw(st.booleans())

schema.update({"type": "boolean", "default": value, "example": value})

elif kind == "array":

schema.update(

{

"type": "array",

"items": {"type": "string", "description": draw(UNTRUSTED_TEXT)},

"example": [draw(UNTRUSTED_TEXT)],

}

)

elif kind == "object":

schema.update(

{

"title": draw(UNTRUSTED_TEXT),

"type": "object",

"properties": {"nested": {"type": "string", "description": draw(UNTRUSTED_TEXT)}},

"additionalProperties": {"type": "string", "description": draw(UNTRUSTED_TEXT)},

"example": {"nested": draw(UNTRUSTED_TEXT)},

}

)

elif kind == "union":

schema = {

"oneOf": [

{"type": "string", "description": description},

{"type": "integer", "description": draw(UNTRUSTED_TEXT)},

],

"description": draw(UNTRUSTED_TEXT),

}

elif kind == "const":

schema = {"const": draw(UNTRUSTED_TEXT), "description": description}

elif kind == "free_form":

schema = {"type": "object", "additionalProperties": True, "description": description}

else:

schema = {"$ref": "#/components/schemas/Choice"}

return schema

@st.composite

def openapi_documents(draw: st.DrawFn) -> dict[str, Any]:

component_name = draw(COMPONENT_NAME)

property_name = draw(NONEMPTY_UNTRUSTED_TEXT)

query_name = draw(NONEMPTY_UNTRUSTED_TEXT)

header_name = draw(NONEMPTY_UNTRUSTED_TEXT)

cookie_name = draw(NONEMPTY_UNTRUSTED_TEXT)

path_segment = draw(NONEMPTY_UNTRUSTED_TEXT)

security_name = draw(NONEMPTY_UNTRUSTED_TEXT)

property_schema = draw(schema_strategy())

parameter_schema = draw(schema_strategy())

enum_values = draw(st.lists(UNTRUSTED_TEXT, min_size=1, max_size=4, unique=True))

int_enum_values = draw(st.lists(st.integers(min_value=-100, max_value=100), min_size=1, max_size=4, unique=True))

return {

"openapi": draw(st.sampled_from(("3.0.0", "3.0.3", "3.1.0"))),

"info": {

"title": draw(UNTRUSTED_TEXT),

"description": draw(UNTRUSTED_TEXT),

"version": draw(UNTRUSTED_TEXT),

"termsOfService": draw(UNTRUSTED_TEXT),

"contact": {

"name": draw(UNTRUSTED_TEXT),

"url": draw(UNTRUSTED_TEXT),

"email": draw(UNTRUSTED_TEXT),

},

"license": {"name": draw(NONEMPTY_UNTRUSTED_TEXT), "url": draw(UNTRUSTED_TEXT)},

},

"servers": [

{

"url": draw(UNTRUSTED_TEXT),

"description": draw(UNTRUSTED_TEXT),

"variables": {

"fuzz": {

"default": draw(UNTRUSTED_TEXT),

"enum": [draw(UNTRUSTED_TEXT)],

"description": draw(UNTRUSTED_TEXT),

}

},

}

],

"tags": [{"name": draw(NONEMPTY_UNTRUSTED_TEXT), "description": draw(UNTRUSTED_TEXT)}],

"externalDocs": {"url": draw(UNTRUSTED_TEXT), "description": draw(UNTRUSTED_TEXT)},

"security": [{security_name: []}],

"paths": {

f"/widgets/{{widget_id}}/{path_segment}": {

"summary": draw(UNTRUSTED_TEXT),

"description": draw(UNTRUSTED_TEXT),

"post": {

"operationId": draw(NONEMPTY_UNTRUSTED_TEXT),

"tags": [draw(UNTRUSTED_TEXT)],

"summary": draw(UNTRUSTED_TEXT),

"description": draw(UNTRUSTED_TEXT),

"parameters": [

{

"name": "widget_id",

"in": "path",

"required": True,

"description": draw(UNTRUSTED_TEXT),

"schema": {"type": "string"},

},

{

"name": query_name,

"in": "query",

"description": draw(UNTRUSTED_TEXT),

"schema": parameter_schema,

},

{

"name": header_name,

"in": "header",

"description": draw(UNTRUSTED_TEXT),

"schema": {"type": "string", "default": draw(UNTRUSTED_TEXT)},

},

{

"name": cookie_name,

"in": "cookie",

"description": draw(UNTRUSTED_TEXT),

"schema": {"type": "string"},

},

],

"requestBody": {

"required": True,

"description": draw(UNTRUSTED_TEXT),

"content": {

"application/json": {

"schema": {"$ref": f"#/components/schemas/{component_name}"},

}

},

},

"responses": {

"200": {

"description": draw(UNTRUSTED_TEXT),

"headers": {

"X-Fuzz": {

"description": draw(UNTRUSTED_TEXT),

"schema": {"type": "string"},

}

},

"content": {

"application/json": {

"schema": {"$ref": f"#/components/schemas/{component_name}"},

}

},

}

},

},

}

},

"components": {

"securitySchemes": {

security_name: {

"type": "apiKey",

"name": draw(NONEMPTY_UNTRUSTED_TEXT),

"in": "header",

"description": draw(UNTRUSTED_TEXT),

}

},

"schemas": {

component_name: {

"title": draw(UNTRUSTED_TEXT),

"type": "object",

"description": draw(UNTRUSTED_TEXT),

"externalDocs": {"url": draw(UNTRUSTED_TEXT), "description": draw(UNTRUSTED_TEXT)},

"xml": {

"name": draw(UNTRUSTED_TEXT),

"namespace": draw(UNTRUSTED_TEXT),

"prefix": draw(UNTRUSTED_TEXT),

},

"required": [property_name],

"properties": {property_name: property_schema},

},

"Choice": {

"type": "string",

"description": draw(UNTRUSTED_TEXT),

"enum": enum_values,

},

"IntChoice": {

"type": "integer",

"description": draw(UNTRUSTED_TEXT),

"enum": int_enum_values,

},

},

},

}

@settings(max_examples=FUZZ_EXAMPLES, deadline=None)

@given(document=openapi_documents(), meta_type=st.sampled_from(list(MetaType)))

def test_valid_openapi_never_generates_invalid_python(document: dict[str, Any], meta_type: MetaType) -> None:

# The generator's own schema model is the source of truth for "valid": anything it accepts

# must produce syntactically valid Python.

OpenAPI.model_validate(document)

with TemporaryDirectory() as temporary_directory:

root = Path(temporary_directory)

document_path = root / "openapi.json"

output_path = root / "generated"

document_path.write_text(json.dumps(document, ensure_ascii=False), encoding="utf-8")

config = Config.from_sources(

ConfigFile(

project_name_override="fuzz-client",

package_name_override="fuzz_client",

post_hooks=[],

),

meta_type,

document_source=document_path,

file_encoding="utf-8",

overwrite=False,

output_path=output_path,

)

with redirect_stdout(io.StringIO()):

errors = generate(config=config)

fatal_errors = [error for error in errors if error.level == ErrorLevel.ERROR]

assert not fatal_errors

python_files = sorted(output_path.rglob("*.py"))

assert python_files

for python_file in python_files:

source = python_file.read_text(encoding="utf-8")

tree = ast.parse(source, filename=str(python_file))

with warnings.catch_warnings():

warnings.simplefilter("error")

compile(source, str(python_file), "exec")

_assert_no_code_injection(tree, python_file)

for pyproject_file in output_path.rglob("pyproject.toml"):

with pyproject_file.open("rb") as pyproject:

tomllib.load(pyproject)

# Static analysis catches escapes which produce valid syntax but reference names that

# don't exist (e.g. an injected function call) — even ones the canary doesn't cover.

ruff = subprocess.run(

["ruff", "check", "--select", "F821", "--no-cache", str(output_path)],

capture_output=True,

text=True,

check=False,

)

assert ruff.returncode == 0, f"Generated code references undefined names:\n{ruff.stdout}"

def _assert_no_code_injection(tree: ast.AST, python_file: Path) -> None:

"""Fail if the canary escaped a string literal/docstring and became executable code."""

for node in ast.walk(tree):

if isinstance(node, ast.Name) and CANARY in node.id:

raise AssertionError(f"Canary escaped into a name in {python_file}: {node.id}")

if isinstance(node, ast.Attribute) and CANARY in node.attr:

raise AssertionError(f"Canary escaped into an attribute in {python_file}: {node.attr}")

if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and CANARY in node.name:

raise AssertionError(f"Canary escaped into a definition in {python_file}: {node.name}")

if isinstance(node, ast.ImportFrom) and node.module and CANARY in node.module:

raise AssertionError(f"Canary escaped into an import in {python_file}: {node.module}")

if isinstance(node, ast.alias) and CANARY in node.name:

raise AssertionError(f"Canary escaped into an import in {python_file}: {node.name}")

if isinstance(node, ast.Call) and any(

isinstance(child, ast.Constant) and child.value == CALL_CANARY for child in ast.walk(node)

):

raise AssertionError(f"Built-in call canary became executable code in {python_file}")