mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-02 07:13:02 +00:00
e8354313aa
Anyone who finds an accidentally published virtual key can now turn it off without an admin. The route takes one key or up to 100, blocks each live one and drops it from every worker's cache, and answers every valid request with the same empty 202. Each source address gets 60 requests an hour, counted in Redis when it is configured so all workers share one count The OpenAPI docs gate now checks the schema of a route's lowest 2xx response instead of only 200, so a route whose success code is 202 can pass it
78 lines
2.7 KiB
Python
Executable file
78 lines
2.7 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""List the POST operations in the proxy's OpenAPI document that Swagger UI cannot explain.
|
|
|
|
Every finding is one line, ``<METHOD> <path> <code>``, so scripts/openapi_docs_gate.py can
|
|
count them per rule the same delta-vs-base way as the other lint gates. The rules:
|
|
|
|
OA001 no request body schema (the handler reads ``request.body()`` and never told FastAPI
|
|
what is in it, see ``inline_request_body`` in custom_openapi_spec.py for the fix)
|
|
OA002 no description (the handler has no docstring and the route no ``description=``)
|
|
OA003 the success response (the lowest 2xx code) has no schema (no ``response_model`` and no
|
|
``responses={200: ...}``)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from collections.abc import Iterator, Mapping
|
|
from typing import Final, NamedTuple
|
|
|
|
GATED_METHODS: Final = frozenset({"post"})
|
|
|
|
|
|
class Violation(NamedTuple):
|
|
method: str
|
|
path: str
|
|
code: str
|
|
|
|
|
|
def _as_mapping(node: object) -> Mapping[str, object]:
|
|
return node if isinstance(node, Mapping) else {}
|
|
|
|
|
|
def _json_schema(container: object) -> object:
|
|
content: Final = _as_mapping(_as_mapping(container).get("content"))
|
|
return _as_mapping(content.get("application/json")).get("schema")
|
|
|
|
|
|
def _success_response(responses: Mapping[str, object]) -> object:
|
|
return next((responses[code] for code in sorted(responses) if code.startswith("2")), None)
|
|
|
|
|
|
def _operation_codes(operation: Mapping[str, object]) -> Iterator[str]:
|
|
if not _json_schema(operation.get("requestBody")):
|
|
yield "OA001"
|
|
description: Final = operation.get("description")
|
|
if not isinstance(description, str) or not description.strip():
|
|
yield "OA002"
|
|
if not _json_schema(_success_response(_as_mapping(operation.get("responses")))):
|
|
yield "OA003"
|
|
|
|
|
|
def violations(spec: Mapping[str, object]) -> tuple[Violation, ...]:
|
|
paths: Final = _as_mapping(spec.get("paths"))
|
|
return tuple(
|
|
Violation(method.upper(), path, code)
|
|
for path, operations in sorted(paths.items())
|
|
for method, operation in sorted(_as_mapping(operations).items())
|
|
if method in GATED_METHODS
|
|
for code in _operation_codes(_as_mapping(operation))
|
|
) # comprehension-ok: one pass over path, method and code is the whole scan
|
|
|
|
|
|
def proxy_spec() -> Mapping[str, object]:
|
|
from litellm.proxy.proxy_server import app
|
|
|
|
return _as_mapping(app.openapi())
|
|
|
|
|
|
def main(argv: list[str] = sys.argv[1:]) -> int:
|
|
spec: Final = json.loads(open(argv[0]).read()) if argv else proxy_spec()
|
|
for violation in violations(spec):
|
|
print(f"{violation.method} {violation.path} {violation.code}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|