replace /proxy route with something more secure

controller.py

@@ -39,10 +39,10 @@ def service_event(event, memo: kopf.Memo, logger, **kwargs):
 
 
     if event['type'] == 'DELETED':
-        if f"{namespace}/{application_name}" in memo.apps:
-            del memo.apps[f"{namespace}/{application_name}"]
+        if f"{namespace}.{application_name}" in memo.apps:
+            del memo.apps[f"{namespace}.{application_name}"]
     else:
-        memo.apps.update({f"{namespace}/{application_name}": {
+        memo.apps.update({f"{namespace}.{application_name}": {
             'url': urlunparse(application_url),
             'name': application_name,
             'header': annotations.get(name_header, ""),
@@ -54,4 +54,7 @@ def service_event(event, memo: kopf.Memo, logger, **kwargs):
     with file_lock:
         with open('static/openapi/urls.json', 'w') as file:
             json.dump(urls, file, indent=2)
-            logger.info("URLs written to file.")
+            logger.info("URLs written to file.")
+        with open('static/openapi/services.json', 'w') as file:
+            json.dump(memo.apps, file, indent=2)
+            logger.info("Services written to file.")

server.py

@@ -6,7 +6,6 @@
 import os
 import requests
 import logging
-from urllib.parse import quote, unquote
 from starlette.responses import RedirectResponse
 from authlib.integrations.starlette_client import OAuth
 from starlette.config import Config
@@ -64,29 +63,29 @@ def require_login(request: Request):
         request.session['user'] = "anonymous"
         return request.session['user']
 
-
-@app.get("/proxy", include_in_schema=False)
-async def proxy(url: str, headers: str = None, user=Depends(require_login)):
-    """
-    Proxy endpoint to fetch the OpenAPI document from a given URL (JSON or YAML).
-    """
-    try:
-        if headers:
-            resp = requests.get(url, headers=json.loads(unquote(headers)), timeout=int(os.environ.get("PROXY_TIMEOUT", 10)))
-        else:
-            resp = requests.get(url, timeout=int(os.environ.get("PROXY_TIMEOUT", 10)))
-        content_type = resp.headers.get("content-type", "")
-        # Se for JSON, repasse como application/json
-        if "json" in content_type:
-            return Response(content=resp.content, media_type="application/json")
-        # Se for YAML, repasse como text/yaml
-        elif "yaml" in content_type or "yml" in content_type:
-            return Response(content=resp.content, media_type="text/yaml")
-        else:
-            raise HTTPException(status_code=400, detail="Unsupported content type")
-    except requests.RequestException as e:
-        logger.error(f"Error fetching OpenAPI document: {e}")
-        raise HTTPException(status_code=500, detail={"error": "Failed to fetch OpenAPI document", "details": str(e)})
+@app.get("/services/{name}", response_class=HTMLResponse, include_in_schema=False)
+async def services(request: Request, name: str, user=Depends(require_login)):
+    with open('static/openapi/services.json', 'r') as f:
+        services = json.load(f)
+        logger.info(f"Loaded {len(services)} services.")
+    if name not in services:
+        logger.error(f"Service {name} not found.")
+        raise HTTPException(status_code=404, detail="Service not found")
+    service = services[name]
+    if not service:
+        logger.error(f"Service {name} not found.")
+        raise HTTPException(status_code=404, detail="Service not found")
+    resp = requests.get(service['url'], timeout=int(os.environ.get("PROXY_TIMEOUT", 10)), headers=parse_headers(service.get("header")))
+    content_type = resp.headers.get("content-type", "")
+    # Se for JSON, repasse como application/json
+    if "json" in content_type:
+        return Response(content=resp.content, media_type="application/json")
+    # Se for YAML, repasse como text/yaml
+    elif "yaml" in content_type or "yml" in content_type:
+        return Response(content=resp.content, media_type="text/yaml")
+    else:
+        logger.error(f"Unsupported content type: {content_type}")
+        raise HTTPException(status_code=400, detail="Unsupported content type")
 
 def parse_headers(header_string: str) -> dict:
     headers = {}
@@ -100,81 +99,59 @@ def parse_headers(header_string: str) -> dict:
     return headers
 
 
-def apply_proxy_to_openapi(openapi_url: str, header: str = None) -> str:
-    """
-    Apply the proxy to the OpenAPI URL.
-    """
-    if openapi_url.startswith("http"):
-        new_url = f"/proxy?url={openapi_url}"
-        if header:
-            header = quote(json.dumps(header))
-            new_url += f"&headers={header}"
-        return new_url
-    return openapi_url
-
-
 @app.get("/", response_class=HTMLResponse)
-async def docs(request: Request, template:str=None, user=Depends(require_login)):
-    """
-    Main documentation page.
-    """
+async def index(request: Request, template:str='swagger-ui', user=Depends(require_login)):
+    if template.lower() not in ["redoc", "swagger-ui"]:
+        raise HTTPException(status_code=400, detail="Invalid template. Use 'redoc' or 'swagger-ui'.")
+
     try:
-        with open('static/openapi/urls.json', 'r') as f:
-            swaggers = json.load(f)
-            logger.info(f"Loaded {len(swaggers)} URLs.")
-    except FileNotFoundError:
-        logger.error("File not found: static/openapi/urls.json")
-        request.session['error'] = "File not found: static/openapi/urls.json"
-        swaggers = [
-            {
-                "url": "/openapi.json",
-                "name": "Swagger Aggregator",
-                "header": "",
-            }
-        ]
-    except json.JSONDecodeError:
-        logger.error("Error decoding JSON from static/openapi/urls.json")
-        request.session['error'] = "Error decoding JSON from static/openapi/urls.json"
-        swaggers = [
-            {
-                "url": "/openapi.json",
-                "name": "Swagger Aggregator",
-                "header": "",
-            }
-        ]
-
-    for swagger in swaggers:
-        swagger["url"] = apply_proxy_to_openapi(swagger.get("url"), parse_headers(swagger.get("header")))
-
-    if template and template.lower() in ["redoc", "swagger-ui"]:
-        return templates.TemplateResponse(
-            f"{template.lower()}.html",
-            {
-                "request": request,
-                "urls": swaggers,
-                "title": os.environ.get("TITLE", "API Documentation"),
-            }
-        )
-    interface = os.environ.get("INTERFACE", "swagger-ui").lower()
-    if interface not in ["swagger-ui", "redoc"]:
-        interface = "swagger-ui"
-    return templates.TemplateResponse(
-        f"{interface}.html",
-        {
-            "request": request,
-            "urls": swaggers,
-            "title": os.environ.get("TITLE", "API Documentation"),
-        }
-    )
+        with open('static/openapi/services.json', 'r') as f:
+            services = json.load(f)
+            logger.info(f"Loaded {len(services)} services.")
+    except Exception as e:
+        logger.error(f"Error loading services file: {e}")
+        raise HTTPException(status_code=500, detail="Error loading services file.")
+
+    urls = []
+    for service_name, service in services.items():
+        urls.append({
+            "url": f"/services/{service_name}",
+            "name": service['name'],
+            "header": service.get("header", ""),
+        })
+
+    match template.lower():
+        case "redoc":
+            return templates.TemplateResponse(
+                "redoc.html",
+                {
+                    "request": request,
+                    "urls": urls,
+                    "title": os.environ.get("TITLE", "API Documentation"),
+                }
+            )
+        case "swagger-ui":
+            return templates.TemplateResponse(
+                "swagger-ui.html",
+                {
+                    "request": request,
+                    "urls": urls,
+                    "title": os.environ.get("TITLE", "API Documentation"),
+                }
+            )
+        case _:
+            logger.error(f"Invalid template: {template}")
+            raise HTTPException(status_code=400, detail="Invalid template. Use 'redoc' or 'swagger-ui'.")
+
 
 @app.get("/config", response_class=HTMLResponse, include_in_schema=False)
 async def config(request: Request, user=Depends(require_login)):
     """
     Configuration page for the OpenAPI URLs.
     """
     try:
-        with open('static/openapi/urls.json') as f:
-            swaggers = json.load(f)
+        with open('static/openapi/services.json') as f:
+            swaggers = json.load(f).values()
     except Exception as e:
         logger.error(f"Error loading configuration file: {e}")
         swaggers = []

tests/test_controller.py

@@ -32,7 +32,7 @@ def test_path_without_slash(fake_memo):
     logger = MagicMock()
     with patch("builtins.open"), patch("json.dump"):
         service_event(event, fake_memo, logger)
-    key = "default/my-app"
+    key = "default.my-app"
     assert key in fake_memo.apps
     # The final path should contain the original path
     assert fake_memo.apps[key]['url'].endswith("/openapi.json")
@@ -43,7 +43,7 @@ def test_path_with_slash(fake_memo):
     logger = MagicMock()
     with patch("builtins.open"), patch("json.dump"):
         service_event(event, fake_memo, logger)
-    key = "default/my-app"
+    key = "default.my-app"
     assert key in fake_memo.apps
     assert fake_memo.apps[key]['url'].endswith("/openapi.json")
 
@@ -53,7 +53,7 @@ def test_path_with_host(fake_memo):
     logger = MagicMock()
     with patch("builtins.open"), patch("json.dump"):
         service_event(event, fake_memo, logger)
-    key = "default/my-app"
+    key = "default.my-app"
     assert key in fake_memo.apps
     # The host should be preserved
     assert fake_memo.apps[key]['url'].startswith("http://myhost")
@@ -72,7 +72,7 @@ def test_missing_annotation(fake_memo):
 
 def test_service_event_deleted(fake_memo):
     # Pre-add an app to memo
-    fake_memo.apps["default/my-app"] = {
+    fake_memo.apps["default.my-app"] = {
         "url": "http://dummy",
         "name": "my-app",
         "header": "X-API-KEY"
@@ -96,4 +96,4 @@ def test_service_event_deleted(fake_memo):
     with patch("builtins.open"), patch("json.dump"):
         service_event(event, fake_memo, logger)
     # Should remove the app from memo
-    assert "default/my-app" not in fake_memo.apps
+    assert "default.my-app" not in fake_memo.apps