437 lines
15 KiB
Python
437 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Build a best-effort OpenAPI manifest from Django DEBUG URL patterns and OPTIONS metadata.
|
|
Discovery-only: sends GET/OPTIONS plus deliberately invalid credential-validation POSTs.
|
|
"""
|
|
|
|
import concurrent.futures
|
|
import html
|
|
import json
|
|
import re
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.request import Request, urlopen
|
|
|
|
BASE = "https://iop-dev.strndev.com"
|
|
OUT = Path(__file__).parent
|
|
UA = "stern-openapi-reconstruction/1.0 (read-only documentation)"
|
|
|
|
|
|
def request(path, method="GET"):
|
|
req = Request(
|
|
BASE + path,
|
|
method=method,
|
|
headers={"Accept": "application/json", "User-Agent": UA},
|
|
)
|
|
try:
|
|
with urlopen(req, timeout=20) as res:
|
|
return res.status, dict(res.headers.items()), res.read()
|
|
except HTTPError as e:
|
|
return e.code, dict(e.headers.items()), e.read()
|
|
except (URLError, OSError, TimeoutError) as e:
|
|
return 0, {}, str(e).encode()
|
|
|
|
|
|
def django_routes(page):
|
|
# Django debug view puts individual patterns in li nodes.
|
|
values = []
|
|
for raw in re.findall(r"<li>(.*?)</li>", page, flags=re.DOTALL):
|
|
value = re.sub(r"\s+", " ", html.unescape(re.sub(r"<[^>]+>", " ", raw))).strip()
|
|
value = re.sub(r"\s*\[name=.*?\]$", "", value)
|
|
if value.startswith(("api/", "webhook/")):
|
|
values.append(value)
|
|
return list(dict.fromkeys(values))
|
|
|
|
|
|
def normalize(route):
|
|
route = route.strip()
|
|
route = route.removeprefix("^").removesuffix("$")
|
|
route = route.replace("?", "")
|
|
# Convert Django path converters to OAS template syntax.
|
|
route = re.sub(
|
|
r"<(?:(?:uuid|slug|str|int):)?([A-Za-z_][A-Za-z0-9_]*)>", r"{\1}", route
|
|
)
|
|
# Never copy externally exposed secret-looking webhook tokens into output.
|
|
if re.search(r"/(?:[A-Za-z0-9]{24,})/?$", route) and "webhook" not in route:
|
|
route = re.sub(r"/[A-Za-z0-9]{24,}/?$", "/{webhook_secret}/", route)
|
|
return "/" + route.lstrip("/")
|
|
|
|
|
|
def options(path):
|
|
status, headers, body = request(path, "OPTIONS")
|
|
lower = {k.lower(): v for k, v in headers.items()}
|
|
try:
|
|
data = json.loads(body)
|
|
except (ValueError, UnicodeDecodeError):
|
|
data = None
|
|
allow = [
|
|
x.strip().lower()
|
|
for x in lower.get("allow", "").split(",")
|
|
if x.strip() and x.strip().upper() not in {"HEAD", "OPTIONS"}
|
|
]
|
|
auth = lower.get("www-authenticate")
|
|
if isinstance(data, dict) and data.get("actions"):
|
|
allow = list(dict.fromkeys(allow + [x.lower() for x in data["actions"]]))
|
|
return {"status": status, "allow": allow, "auth": auth, "metadata": data}
|
|
|
|
|
|
def merge_shapes(shapes: list[dict[str, Any]]) -> dict[str, Any]:
|
|
if not shapes:
|
|
return {}
|
|
types = {shape.get("type") for shape in shapes}
|
|
if len(types) != 1:
|
|
unique = {json.dumps(shape, sort_keys=True): shape for shape in shapes}
|
|
return {"oneOf": list(unique.values())}
|
|
result = dict(shapes[0])
|
|
if result.get("type") == "object":
|
|
keys = set().union(*(shape.get("properties", {}) for shape in shapes))
|
|
result["properties"] = {
|
|
key: merge_shapes(
|
|
[
|
|
shape["properties"][key]
|
|
for shape in shapes
|
|
if key in shape.get("properties", {})
|
|
]
|
|
)
|
|
for key in sorted(keys)
|
|
}
|
|
return result
|
|
|
|
|
|
def json_shape(value: Any) -> dict[str, Any]:
|
|
if isinstance(value, dict):
|
|
return {
|
|
"type": "object",
|
|
"properties": {str(key): json_shape(item) for key, item in value.items()},
|
|
}
|
|
if isinstance(value, list):
|
|
return {
|
|
"type": "array",
|
|
"items": merge_shapes([json_shape(item) for item in value]),
|
|
}
|
|
if isinstance(value, bool):
|
|
return {"type": "boolean"}
|
|
if isinstance(value, int):
|
|
return {"type": "integer"}
|
|
if isinstance(value, float):
|
|
return {"type": "number"}
|
|
if value is None:
|
|
return {"nullable": True}
|
|
return {"type": "string"}
|
|
|
|
|
|
def schema_example(schema: dict[str, Any]) -> Any:
|
|
if "oneOf" in schema:
|
|
return schema_example(schema["oneOf"][0])
|
|
if schema.get("type") == "object":
|
|
return {
|
|
name: schema_example(value)
|
|
for name, value in schema.get("properties", {}).items()
|
|
}
|
|
if schema.get("type") == "array":
|
|
return [schema_example(schema.get("items", {}))]
|
|
primitive_examples = {
|
|
"boolean": False,
|
|
"integer": 0,
|
|
"number": 0,
|
|
"string": "string",
|
|
}
|
|
schema_type = schema.get("type")
|
|
return primitive_examples.get(schema_type) if isinstance(schema_type, str) else None
|
|
|
|
|
|
def post_validation_evidence(path: str) -> dict[str, Any]:
|
|
req = Request(
|
|
BASE + path,
|
|
data=b"{}",
|
|
method="POST",
|
|
headers={
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
"User-Agent": UA,
|
|
},
|
|
)
|
|
try:
|
|
with urlopen(req, timeout=20) as res:
|
|
status, headers, body = res.status, dict(res.headers.items()), res.read()
|
|
except HTTPError as exc:
|
|
status, headers, body = exc.code, dict(exc.headers.items()), exc.read()
|
|
except (URLError, OSError, TimeoutError) as exc:
|
|
return {"status": 0, "error": str(exc)}
|
|
result: dict[str, Any] = {
|
|
"status": status,
|
|
"contentType": headers.get("Content-Type", ""),
|
|
"bytes": len(body),
|
|
}
|
|
if "json" in result["contentType"].lower():
|
|
try:
|
|
result["schema"] = json_shape(json.loads(body))
|
|
except (ValueError, UnicodeDecodeError):
|
|
result["jsonParseError"] = True
|
|
return result
|
|
|
|
|
|
def get_evidence(path: str) -> dict[str, Any]:
|
|
status, headers, body = request(path)
|
|
content_type = headers.get("Content-Type", "")
|
|
result: dict[str, Any] = {
|
|
"status": status,
|
|
"contentType": content_type,
|
|
"bytes": len(body),
|
|
}
|
|
if 200 <= status < 300 and "json" in content_type.lower():
|
|
try:
|
|
result["schema"] = json_shape(json.loads(body))
|
|
except (ValueError, UnicodeDecodeError):
|
|
result["jsonParseError"] = True
|
|
return result
|
|
|
|
|
|
def field_schema(fields: dict[str, Any]) -> dict[str, Any]:
|
|
props: dict[str, Any] = {}
|
|
required: list[str] = []
|
|
mapping = {
|
|
"string": "string",
|
|
"integer": "integer",
|
|
"boolean": "boolean",
|
|
"number": "number",
|
|
"field": "object",
|
|
"choice": "string",
|
|
"file upload": "string",
|
|
}
|
|
for name, field in fields.items():
|
|
if not isinstance(field, dict):
|
|
continue
|
|
schema: dict[str, Any] = {
|
|
"type": mapping.get(field.get("type", "string"), "string")
|
|
}
|
|
if field.get("label"):
|
|
schema["title"] = field["label"]
|
|
if field.get("read_only"):
|
|
schema["readOnly"] = True
|
|
if field.get("write_only"):
|
|
schema["writeOnly"] = True
|
|
if field.get("choices"):
|
|
schema["enum"] = [
|
|
c.get("value", c) if isinstance(c, dict) else c
|
|
for c in field["choices"]
|
|
]
|
|
props[name] = schema
|
|
if field.get("required"):
|
|
required.append(name)
|
|
result = {"type": "object", "properties": props}
|
|
if required:
|
|
result["required"] = required
|
|
return result
|
|
|
|
|
|
def path_parameters(path):
|
|
return [
|
|
{
|
|
"name": name,
|
|
"in": "path",
|
|
"required": True,
|
|
"schema": {
|
|
"type": "string",
|
|
"format": "uuid" if name in {"audit_download_uid", "uuid"} else None,
|
|
},
|
|
}
|
|
for name in re.findall(r"\{([A-Za-z_][A-Za-z0-9_]*)\}", path)
|
|
]
|
|
|
|
|
|
def operation(
|
|
method: str,
|
|
evidence: dict[str, Any],
|
|
get_result: dict[str, Any] | None,
|
|
validation_result: dict[str, Any] | None,
|
|
) -> dict[str, Any]:
|
|
metadata, auth = evidence.get("metadata"), evidence.get("auth")
|
|
success: dict[str, Any] = {"description": "Success (shape not fully inferred)"}
|
|
if method == "get" and get_result and "schema" in get_result:
|
|
success = {
|
|
"description": "Observed public response shape; field optionality and array item variance are not inferred.",
|
|
"content": {
|
|
"application/json": {
|
|
"schema": get_result["schema"],
|
|
"example": schema_example(get_result["schema"]),
|
|
}
|
|
},
|
|
}
|
|
op: dict[str, Any] = {
|
|
"responses": {
|
|
"200": success,
|
|
"401": {"$ref": "#/components/responses/Unauthorized"},
|
|
"403": {"$ref": "#/components/responses/Forbidden"},
|
|
}
|
|
}
|
|
if method == "post" and validation_result and "schema" in validation_result:
|
|
op["responses"][str(validation_result["status"])] = {
|
|
"description": "Observed validation response to an empty JSON object.",
|
|
"content": {
|
|
"application/json": {
|
|
"schema": validation_result["schema"],
|
|
"example": schema_example(validation_result["schema"]),
|
|
}
|
|
},
|
|
}
|
|
if isinstance(metadata, dict):
|
|
if metadata.get("name"):
|
|
op["summary"] = metadata["name"]
|
|
if metadata.get("description"):
|
|
op["description"] = metadata["description"]
|
|
fields = metadata.get("actions", {}).get(method.upper())
|
|
if isinstance(fields, dict):
|
|
op["requestBody"] = {
|
|
"required": True,
|
|
"content": {
|
|
ctype: {
|
|
"schema": field_schema(fields),
|
|
"example": schema_example(field_schema(fields)),
|
|
}
|
|
for ctype in metadata.get("parses", ["application/json"])
|
|
},
|
|
}
|
|
if auth:
|
|
op["security"] = [{"bearerAuth": []}]
|
|
return op
|
|
|
|
|
|
def main():
|
|
status, _, debug = request("/openapi-reconstruction-does-not-exist")
|
|
if status != 404:
|
|
raise SystemExit(f"Expected Django debug 404; received {status}")
|
|
page = debug.decode("utf-8", "replace")
|
|
raw_routes = django_routes(page)
|
|
paths = sorted({normalize(route) for route in raw_routes})
|
|
# Retain only normalized paths: the raw DEBUG page contains opaque webhook URLs.
|
|
(OUT / "django-404-urlpatterns.routes.txt").write_text("\n".join(paths) + "\n")
|
|
|
|
# OPTIONS is a safe discovery method but is throttled to five concurrent requests.
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as pool:
|
|
records = dict(zip(paths, pool.map(options, paths), strict=True))
|
|
(OUT / "endpoint-options-evidence.json").write_text(
|
|
json.dumps(records, indent=2, sort_keys=True, default=str)
|
|
)
|
|
readable_paths = [
|
|
path for path, record in records.items() if "get" in record["allow"]
|
|
]
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as pool:
|
|
get_records = dict(
|
|
zip(readable_paths, pool.map(get_evidence, readable_paths), strict=True)
|
|
)
|
|
(OUT / "endpoint-get-evidence.json").write_text(
|
|
json.dumps(get_records, indent=2, sort_keys=True)
|
|
)
|
|
# These endpoints only validate credentials/registration data; `{}` is deliberately invalid and cannot complete their action.
|
|
validation_paths = [
|
|
"/api/v2/token/",
|
|
"/api/v2/portal/user_verify_email/",
|
|
"/api/v2/portal/user_verify_password/",
|
|
"/api/v2/portal/user_verify_username/",
|
|
"/api/v2/portal/user_confirm_registration/",
|
|
"/api/v3/portal/user_auth/",
|
|
]
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
|
|
validation_records = dict(
|
|
zip(
|
|
validation_paths,
|
|
pool.map(post_validation_evidence, validation_paths),
|
|
strict=True,
|
|
)
|
|
)
|
|
(OUT / "endpoint-validation-evidence.json").write_text(
|
|
json.dumps(validation_records, indent=2, sort_keys=True)
|
|
)
|
|
|
|
manifest = {
|
|
"openapi": "3.0.3",
|
|
"info": {
|
|
"title": "Stern Insider (iop-dev) reconstructed API",
|
|
"version": "0.1.0-reconstructed",
|
|
"description": "Best-effort manifest reconstructed from the public Django DEBUG URLconf and read-only OPTIONS responses. It is not an authoritative contract; operations without OPTIONS metadata have inferred methods only.",
|
|
},
|
|
"servers": [{"url": BASE}],
|
|
"paths": {},
|
|
"components": {
|
|
"securitySchemes": {
|
|
"bearerAuth": {
|
|
"type": "http",
|
|
"scheme": "bearer",
|
|
"bearerFormat": "JWT",
|
|
}
|
|
},
|
|
"responses": {
|
|
"Unauthorized": {
|
|
"description": "Authentication credentials were not provided or are invalid."
|
|
},
|
|
"Forbidden": {"description": "Authenticated caller lacks permission."},
|
|
},
|
|
},
|
|
"x-reconstruction": {
|
|
"generatedAt": datetime.now(timezone.utc).isoformat(),
|
|
"source": "Public Django DEBUG 404 URLconf plus OPTIONS metadata",
|
|
"routeCount": len(paths),
|
|
"rawRouteCount": len(raw_routes),
|
|
"limitations": [
|
|
"Response schemas were not inferred unless supplied by the server.",
|
|
"OPTIONS may be denied before serializer metadata is exposed.",
|
|
"Secret-looking webhook path tokens are redacted as {webhook_secret}.",
|
|
"Non-api HTML routes are intentionally excluded; see API_RECONSTRUCTION.md.",
|
|
],
|
|
},
|
|
}
|
|
for path in paths:
|
|
evidence = records[path]
|
|
methods = evidence["allow"]
|
|
item: dict[str, Any] = {
|
|
m: operation(
|
|
m, evidence, get_records.get(path), validation_records.get(path)
|
|
)
|
|
for m in methods
|
|
}
|
|
parameters = path_parameters(path)
|
|
for parameter in parameters:
|
|
parameter["schema"] = {
|
|
key: value
|
|
for key, value in parameter["schema"].items()
|
|
if value is not None
|
|
}
|
|
if parameters:
|
|
item["parameters"] = parameters
|
|
item["x-options-evidence"] = {
|
|
"status": evidence["status"],
|
|
"auth": evidence["auth"],
|
|
"hasSerializerMetadata": bool(
|
|
isinstance(evidence["metadata"], dict)
|
|
and evidence["metadata"].get("actions")
|
|
),
|
|
}
|
|
if not methods:
|
|
item["x-reconstruction-status"] = (
|
|
"Route is exposed in the Django URLconf, but no operation was inferred because OPTIONS returned no Allow header."
|
|
)
|
|
manifest["paths"][path] = item
|
|
(OUT / "openapi.reconstructed.json").write_text(
|
|
json.dumps(manifest, indent=2) + "\n"
|
|
)
|
|
http_methods = {"get", "post", "put", "patch", "delete", "head", "options", "trace"}
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"routes": len(paths),
|
|
"operationsDocumented": sum(
|
|
len([k for k in p if k in http_methods])
|
|
for p in manifest["paths"].values()
|
|
),
|
|
"pathsWithOperations": len(manifest["paths"]),
|
|
},
|
|
indent=2,
|
|
)
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|