swadhin / justapi public

swadhinbiswas

[email protected]
2026-07-24 19:03:15 +0600 committed NaN years ago
f1875db
commit f1875db
1 file changed
+53 -10
Showing 1 changed file+53-10
crates/justapi-py/python/justapi/app.py
▼crates/justapi-py/python/justapi/app.py
+53-10
@@ -46,7 +46,10 @@ def _builtin_ready(self):
46
46
return JSONResponse(payload)
47
47
48
48
49
-def _builtin_metrics(self):
49
+def _builtin_metrics(self, request=None):
50
+ guard = _admin_guard(self, request)
51
+ if guard is not None:
52
+ return guard
50
53
body = self._app.metrics_prometheus()
51
54
return PlainTextResponse(body)
52
55
@@ -69,13 +72,19 @@ def _builtin_graphql(app, request=None):
69
72
70
73
71
74
72
-def _builtin_openapi(app):
75
+def _builtin_openapi(app, request=None):
73
76
"""Serve the generated OpenAPI 3.1 document for the Python app."""
77
+ guard = _admin_guard(app, request)
78
+ if guard is not None:
79
+ return guard
74
80
return JSONResponse(build_openapi(app))
75
81
76
82
77
-def _builtin_docs(app):
83
+def _builtin_docs(app, request=None):
78
84
"""Serve Swagger UI (interactive API documentation) for the Python app."""
85
+ guard = _admin_guard(app, request)
86
+ if guard is not None:
87
+ return guard
79
88
html = (
80
89
"<!DOCTYPE html>\n<html lang=\"en\">\n<head>\n"
81
90
" <meta charset=\"UTF-8\">\n <title>API Docs — Swagger UI</title>\n"
@@ -91,8 +100,11 @@ def _builtin_docs(app):
91
100
return HTMLResponse(html)
92
101
93
102
94
-def _builtin_redoc(app):
103
+def _builtin_redoc(app, request=None):
95
104
"""Serve ReDoc (alternative API documentation) for the Python app."""
105
+ guard = _admin_guard(app, request)
106
+ if guard is not None:
107
+ return guard
96
108
html = (
97
109
"<!DOCTYPE html>\n<html lang=\"en\">\n<head>\n"
98
110
" <meta charset=\"UTF-8\">\n <title>API Docs — ReDoc</title>\n"
@@ -104,7 +116,7 @@ def _builtin_redoc(app):
104
116
return HTMLResponse(html)
105
117
106
118
107
-def _builtin_scalar(app):
119
+def _builtin_scalar(app, request=None):
108
120
"""Serve Scalar API Reference (modern interactive docs) for the Python app.
109
121
110
122
Scalar reads the OpenAPI document from ``/openapi.json`` and renders a
@@ -239,6 +251,31 @@ class _AppState:
239
251
self.__dict__[name] = value
240
252
241
253
254
+def _admin_guard(app, request):
255
+ """Check admin token if configured.
256
+
257
+ Returns a 403 JSONResponse when the token is set and the request does
258
+ not carry ``Authorization: Bearer <token>``, or ``None`` when authorized.
259
+ """
260
+ token = getattr(app, 'admin_token', None)
261
+ if token is None:
262
+ return None
263
+ auth_header = ""
264
+ if isinstance(request, dict):
265
+ hdrs = request.get("headers") or {}
266
+ if isinstance(hdrs, dict):
267
+ auth_header = hdrs.get("Authorization", "")
268
+ elif isinstance(hdrs, list):
269
+ for k, v in hdrs:
270
+ if k.lower() == b"authorization":
271
+ auth_header = v if isinstance(v, str) else v.decode("utf-8", errors="replace")
272
+ break
273
+ if not auth_header.startswith("Bearer ") or auth_header[7:] != token:
274
+ from .responses import JSONResponse
275
+ return JSONResponse({"detail": "Forbidden"}, status_code=403)
276
+ return None
277
+
278
+
242
279
class JustAPIApp:
243
280
def __init__(
244
281
self,
@@ -255,6 +292,7 @@ class JustAPIApp:
255
292
docs_url: str = "/docs",
256
293
redoc_url: str = "/redoc",
257
294
scalar_url: str = "/scalar",
295
+ admin_token: str = None,
258
296
):
259
297
self._app = _JustAPIApp()
260
298
self.title = title
@@ -277,22 +315,27 @@ class JustAPIApp:
277
315
self._startup_handlers = []
278
316
self._shutdown_handlers = []
279
317
self.state = _AppState()
318
+ self.admin_token = admin_token
280
319
281
320
282
321
# Built-in probe/metrics endpoints for the Python app.
322
+ # Health/live/ready are deliberately unprotected (K8s probes).
283
323
self.get("/health", _builtin_health, include_in_schema=False, name="builtin_health")
284
324
self.get("/live", _builtin_live, include_in_schema=False, name="builtin_live")
285
325
self.get("/ready", lambda: _builtin_ready(self), include_in_schema=False, name="builtin_ready")
286
- self.get("/metrics", lambda: _builtin_metrics(self), include_in_schema=False, name="builtin_metrics")
326
+ self.get("/metrics", lambda request=None: _builtin_metrics(self, request), include_in_schema=False, name="builtin_metrics")
287
327
# Interactive API documentation for the Python app. The core server's
288
328
# /docs, /redoc, /openapi.json live only in the default router, which the
289
329
# Python app replaces via with_handler(); these builtin routes restore
290
330
# that DX so the README's "automatic interactive API documentation"
291
331
# claim holds for Python apps too.
292
- self.get("/openapi.json", lambda: _builtin_openapi(self), include_in_schema=False, name="builtin_openapi")
293
- self.get("/docs", lambda: _builtin_docs(self), include_in_schema=False, name="builtin_docs")
294
- self.get("/redoc", lambda: _builtin_redoc(self), include_in_schema=False, name="builtin_redoc")
295
- self.get("/scalar", lambda: _builtin_scalar(self), include_in_schema=False, name="builtin_scalar")
332
+ # Admin-protected: when `admin_token` is set, the request must carry
333
+ # `Authorization: Bearer <token>` to access these routes.
334
+ self.get("/openapi.json", lambda request=None: _builtin_openapi(self, request), include_in_schema=False, name="builtin_openapi")
335
+ self.get("/docs", lambda request=None: _builtin_docs(self, request), include_in_schema=False, name="builtin_docs")
336
+ self.get("/redoc", lambda request=None: _builtin_redoc(self, request), include_in_schema=False, name="builtin_redoc")
337
+ # The scalar docs page is also protected; the home page (/) is not.
338
+ self.get("/scalar", lambda request=None: _admin_guard(self, request) or _builtin_scalar(self, request), include_in_schema=False, name="builtin_scalar")
296
339
# The home page opens the interactive API reference (Scalar) so the
297
340
# docs are the first thing a visitor sees.
298
341
self.get("/", lambda: _builtin_scalar(self), include_in_schema=False, name="builtin_index")
299
342