Pycturator should allow disabling specifc endpoints

In order to disable pyctuator endpoints such as `/pyctuator/httptraces`, pyctuator now
can be initialized with flags selecting endpoints that shouldn't be registered in SBA and
should fail if queried (GET) directly.

Fixes #92

pyctuator/endpoints.py

@@ -0,0 +1,13 @@
+from enum import Flag, auto
+
+
+class Endpoints(Flag):
+    NONE = 0
+    ENV = auto()
+    INFO = auto()
+    HEALTH = auto()
+    METRICS = auto()
+    LOGGERS = auto()
+    THREAD_DUMP = auto()
+    LOGFILE = auto()
+    HTTP_TRACE = auto()

pyctuator/impl/aiohttp_pyctuator.py

@@ -9,6 +9,7 @@
 from aiohttp import web
 from multidict import CIMultiDictProxy
 
+from pyctuator.endpoints import Endpoints
 from pyctuator.httptrace import TraceRecord, TraceRequest, TraceResponse
 from pyctuator.impl import SBA_V2_CONTENT_TYPE
 from pyctuator.impl.pyctuator_impl import PyctuatorImpl
@@ -17,7 +18,7 @@
 
 # pylint: disable=too-many-locals,unused-argument
 class AioHttpPyctuator(PyctuatorRouter):
-    def __init__(self, app: web.Application, pyctuator_impl: PyctuatorImpl) -> None:
+    def __init__(self, app: web.Application, pyctuator_impl: PyctuatorImpl, disabled_endpoints: Endpoints) -> None:
         super().__init__(app, pyctuator_impl)
 
         custom_dumps = partial(
@@ -106,34 +107,50 @@ async def intercept_requests_and_responses(request: web.Request, handler: Callab
             self.pyctuator_impl.http_tracer.add_record(record=new_record)
             return response
 
-        app.add_routes(
-            [
-                web.get("/pyctuator", get_endpoints),
-                web.options("/pyctuator/env", empty_handler),
-                web.options("/pyctuator/info", empty_handler),
-                web.options("/pyctuator/health", empty_handler),
-                web.options("/pyctuator/metrics", empty_handler),
-                web.options("/pyctuator/loggers", empty_handler),
-                web.options("/pyctuator/dump", empty_handler),
-                web.options("/pyctuator/threaddump", empty_handler),
-                web.options("/pyctuator/logfile", empty_handler),
-                web.options("/pyctuator/trace", empty_handler),
-                web.options("/pyctuator/httptrace", empty_handler),
-                web.get("/pyctuator/env", get_environment),
-                web.get("/pyctuator/info", get_info),
-                web.get("/pyctuator/health", get_health),
-                web.get("/pyctuator/metrics", get_metric_names),
-                web.get("/pyctuator/metrics/{metric_name}", get_metric_measurement),
-                web.get("/pyctuator/loggers", get_loggers),
-                web.get("/pyctuator/loggers/{logger_name}", get_logger),
-                web.post("/pyctuator/loggers/{logger_name}", set_logger_level),
-                web.get("/pyctuator/dump", get_thread_dump),
-                web.get("/pyctuator/threaddump", get_thread_dump),
-                web.get("/pyctuator/logfile", get_logfile),
-                web.get("/pyctuator/trace", get_httptrace),
-                web.get("/pyctuator/httptrace", get_httptrace),
-            ]
-        )
+        routes = [
+            web.get("/pyctuator", get_endpoints),
+        ]
+
+        if Endpoints.ENV not in disabled_endpoints:
+            routes.append(web.options("/pyctuator/env", empty_handler))
+            routes.append(web.get("/pyctuator/env", get_environment))
+
+        if Endpoints.INFO not in disabled_endpoints:
+            routes.append(web.options("/pyctuator/info", empty_handler))
+            routes.append(web.get("/pyctuator/info", get_info))
+
+        if Endpoints.HEALTH not in disabled_endpoints:
+            routes.append(web.options("/pyctuator/health", empty_handler))
+            routes.append(web.get("/pyctuator/health", get_health))
+
+        if Endpoints.METRICS not in disabled_endpoints:
+            routes.append(web.options("/pyctuator/metrics", empty_handler))
+            routes.append(web.get("/pyctuator/metrics", get_metric_names))
+            routes.append(web.get("/pyctuator/metrics/{metric_name}", get_metric_measurement))
+
+        if Endpoints.LOGGERS not in disabled_endpoints:
+            routes.append(web.options("/pyctuator/loggers", empty_handler))
+            routes.append(web.get("/pyctuator/loggers", get_loggers))
+            routes.append(web.get("/pyctuator/loggers/{logger_name}", get_logger))
+            routes.append(web.post("/pyctuator/loggers/{logger_name}", set_logger_level))
+
+        if Endpoints.THREAD_DUMP not in disabled_endpoints:
+            routes.append(web.options("/pyctuator/dump", empty_handler))
+            routes.append(web.options("/pyctuator/threaddump", empty_handler))
+            routes.append(web.get("/pyctuator/dump", get_thread_dump))
+            routes.append(web.get("/pyctuator/threaddump", get_thread_dump))
+
+        if Endpoints.LOGFILE not in disabled_endpoints:
+            routes.append(web.options("/pyctuator/logfile", empty_handler))
+            routes.append(web.get("/pyctuator/logfile", get_logfile))
+
+        if Endpoints.HTTP_TRACE not in disabled_endpoints:
+            routes.append(web.options("/pyctuator/trace", empty_handler))
+            routes.append(web.options("/pyctuator/httptrace", empty_handler))
+            routes.append(web.get("/pyctuator/trace", get_httptrace))
+            routes.append(web.get("/pyctuator/httptrace", get_httptrace))
+
+        app.add_routes(routes)
         app.middlewares.append(intercept_requests_and_responses)
 
     def _custom_json_serializer(self, value: Any) -> Any:

pyctuator/impl/fastapi_pyctuator.py

@@ -10,6 +10,7 @@
 from starlette.requests import Request
 from starlette.responses import Response
 
+from pyctuator.endpoints import Endpoints
 from pyctuator.environment.environment_provider import EnvironmentData
 from pyctuator.httptrace import TraceRecord, TraceRequest, TraceResponse
 from pyctuator.httptrace.http_tracer import Traces
@@ -33,8 +34,9 @@ def __init__(
             self,
             app: FastAPI,
             pyctuator_impl: PyctuatorImpl,
-            include_in_openapi_schema: bool = False,
-            customizer: Optional[Callable[[APIRouter], None]] = None
+            include_in_openapi_schema: bool,
+            customizer: Optional[Callable[[APIRouter], None]],
+            disabled_endpoints: Endpoints,
     ) -> None:
         super().__init__(app, pyctuator_impl)
         router = APIRouter()
@@ -64,70 +66,78 @@ def options() -> None:
             documentation.
             """
 
-        @router.get("/env", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
-        def get_environment() -> EnvironmentData:
-            return pyctuator_impl.get_environment()
+        if Endpoints.ENV not in disabled_endpoints:
+            @router.get("/env", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
+            def get_environment() -> EnvironmentData:
+                return pyctuator_impl.get_environment()
 
-        @router.get("/info", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
-        def get_info() -> Dict:
-            return pyctuator_impl.get_app_info()
+        if Endpoints.INFO not in disabled_endpoints:
+            @router.get("/info", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
+            def get_info() -> Dict:
+                return pyctuator_impl.get_app_info()
 
-        @router.get("/health", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
-        def get_health(response: Response) -> object:
-            health = pyctuator_impl.get_health()
-            response.status_code = health.http_status()
-            return health
+        if Endpoints.HEALTH not in disabled_endpoints:
+            @router.get("/health", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
+            def get_health(response: Response) -> object:
+                health = pyctuator_impl.get_health()
+                response.status_code = health.http_status()
+                return health
 
-        @router.get("/metrics", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
-        def get_metric_names() -> MetricNames:
-            return pyctuator_impl.get_metric_names()
+        if Endpoints.METRICS not in disabled_endpoints:
+            @router.get("/metrics", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
+            def get_metric_names() -> MetricNames:
+                return pyctuator_impl.get_metric_names()
 
-        @router.get("/metrics/{metric_name}", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
-        def get_metric_measurement(metric_name: str) -> Metric:
-            return pyctuator_impl.get_metric_measurement(metric_name)
+            @router.get("/metrics/{metric_name}", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
+            def get_metric_measurement(metric_name: str) -> Metric:
+                return pyctuator_impl.get_metric_measurement(metric_name)
 
         # Retrieving All Loggers
-        @router.get("/loggers", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
-        def get_loggers() -> LoggersData:
-            return pyctuator_impl.logging.get_loggers()
-
-        @router.post("/loggers/{logger_name}", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
-        def set_logger_level(item: FastApiLoggerItem, logger_name: str) -> Dict:
-            pyctuator_impl.logging.set_logger_level(logger_name, item.configuredLevel)
-            return {}
-
-        @router.get("/loggers/{logger_name}", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
-        def get_logger(logger_name: str) -> LoggerLevels:
-            return pyctuator_impl.logging.get_logger(logger_name)
-
-        @router.get("/dump", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
-        @router.get("/threaddump", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
-        def get_thread_dump() -> ThreadDump:
-            return pyctuator_impl.get_thread_dump()
-
-        @router.get("/logfile", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
-        def get_logfile(range_header: str = Header(default=None,
-                                                   alias="range")) -> Response:  # pylint: disable=redefined-builtin
-            if not range_header:
-                return Response(content=pyctuator_impl.logfile.log_messages.get_range())
-
-            str_res, start, end = pyctuator_impl.logfile.get_logfile(range_header)
-
-            my_res = Response(
-                status_code=HTTPStatus.PARTIAL_CONTENT.value,
-                content=str_res,
-                headers={
-                    "Content-Type": "text/html; charset=UTF-8",
-                    "Accept-Ranges": "bytes",
-                    "Content-Range": f"bytes {start}-{end}/{end}",
-                })
-
-            return my_res
-
-        @router.get("/trace", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
-        @router.get("/httptrace", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
-        def get_httptrace() -> Traces:
-            return pyctuator_impl.http_tracer.get_httptrace()
+        if Endpoints.LOGGERS not in disabled_endpoints:
+            @router.get("/loggers", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
+            def get_loggers() -> LoggersData:
+                return pyctuator_impl.logging.get_loggers()
+
+            @router.post("/loggers/{logger_name}", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
+            def set_logger_level(item: FastApiLoggerItem, logger_name: str) -> Dict:
+                pyctuator_impl.logging.set_logger_level(logger_name, item.configuredLevel)
+                return {}
+
+            @router.get("/loggers/{logger_name}", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
+            def get_logger(logger_name: str) -> LoggerLevels:
+                return pyctuator_impl.logging.get_logger(logger_name)
+
+        if Endpoints.THREAD_DUMP not in disabled_endpoints:
+            @router.get("/dump", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
+            @router.get("/threaddump", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
+            def get_thread_dump() -> ThreadDump:
+                return pyctuator_impl.get_thread_dump()
+
+        if Endpoints.LOGFILE not in disabled_endpoints:
+            @router.get("/logfile", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
+            def get_logfile(range_header: str = Header(default=None,
+                                                       alias="range")) -> Response:  # pylint: disable=redefined-builtin
+                if not range_header:
+                    return Response(content=pyctuator_impl.logfile.log_messages.get_range())
+
+                str_res, start, end = pyctuator_impl.logfile.get_logfile(range_header)
+
+                my_res = Response(
+                    status_code=HTTPStatus.PARTIAL_CONTENT.value,
+                    content=str_res,
+                    headers={
+                        "Content-Type": "text/html; charset=UTF-8",
+                        "Accept-Ranges": "bytes",
+                        "Content-Range": f"bytes {start}-{end}/{end}",
+                    })
+
+                return my_res
+
+        if Endpoints.HTTP_TRACE not in disabled_endpoints:
+            @router.get("/trace", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
+            @router.get("/httptrace", include_in_schema=include_in_openapi_schema, tags=["pyctuator"])
+            def get_httptrace() -> Traces:
+                return pyctuator_impl.http_tracer.get_httptrace()
 
         @app.middleware("http")
         async def intercept_requests_and_responses(