github/KubeDiagrams
1
0
Fork 0
mirror of https://github.com/philippemerle/KubeDiagrams.git synced 2026-08-16 14:16:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(webapp) : harden backend input validation and error handling

Browse Source
This commit is contained in:
Sadallah
2026-07-17 00:59:05 +02:00
parent 637b7786b4
commit 3d05ce9ada
19 changed files with 309 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
webapp/backend/config.py
+1 -1
View File
@@ -8,7 +8,7 @@ class Config:
"""Config App and Logger."""
# Flask
DEBUG = True
DEBUG = os.environ.get('FLASK_DEBUG', 'false').lower() == 'true'
PORT = 5000
HOST = 'localhost'
webapp/backend/constants.py
+20
View File
@@ -25,3 +25,23 @@ MAX_LOG_LENGTH = 999999
# file extensions
YAML_EXTENSIONS = ['.yaml', '.yml']
TGZ_EXTENSIONS = ['.tgz', '.tar.gz']
# Allowlist of CLI flags accepted through the free-text "extra args" field, per
# underlying tool. -o/--output, -f/--format and -c/--config are excluded even
# though the tools support them: -o/-f are already managed by the app itself
# (allowing them would let a request override the computed output path/format),
# and -c/--config (plus helm's --values/--set-file) let the tool read an
# arbitrary local file path, which would be a local file disclosure primitive.
EXTRA_ARGS_ALLOWED_FLAGS = {
"kube-diagrams": {
"--embed-all-icons", "-v", "--verbose", "-n", "--namespace", "--without-namespace",
},
"kubectl-diagrams": {
"--embed-all-icons", "--version",
},
"helm-diagrams": {
"--set", "--set-string", "--set-json", "--set-literal",
"-g", "--generate-name", "--include-crds", "-l", "--labels",
"--name-template", "--version", "--embed-all-icons",
},
}
webapp/backend/routes/cluster.py
+19 -28
View File
@@ -16,43 +16,34 @@ cluster_bp = Blueprint('cluster', __name__)
@cluster_bp.route('/api/cluster/context', methods=['GET'])
def get_context():
"""Return the name of the currently active kubectl context."""
try:
context = get_current_context()
return ResponseBuilder.success({"context": context})
except RuntimeError as e:
return ResponseBuilder.error(str(e))
except Exception as e:
return ResponseBuilder.error(f"Unexpected error: {str(e)}")
context, error = get_current_context()
if error:
return ResponseBuilder.error(error)
return ResponseBuilder.success({"context": context})
@cluster_bp.route('/api/cluster/namespaces', methods=['GET'])
def list_namespaces():
"""Return the list of namespaces available in the connected Kubernetes cluster."""
try:
namespaces = get_namespaces()
return ResponseBuilder.success({
"namespaces": namespaces,
"count": len(namespaces)
})
except RuntimeError as e:
return ResponseBuilder.error(str(e))
except Exception as e:
return ResponseBuilder.error(f"Unexpected error: {str(e)}")
namespaces, error = get_namespaces()
if error:
return ResponseBuilder.error(error)
return ResponseBuilder.success({
"namespaces": namespaces,
"count": len(namespaces)
})
@cluster_bp.route('/api/cluster/resource-types', methods=['GET'])
def list_resource_types():
"""Return all resource types known by the cluster, tagged with namespace scope and common status."""
try:
resource_types = get_resource_types()
return ResponseBuilder.success({
"resourceTypes": resource_types,
"count": len(resource_types)
})
except RuntimeError as e:
return ResponseBuilder.error(str(e))
except Exception as e:
return ResponseBuilder.error(f"Unexpected error: {str(e)}")
resource_types, error = get_resource_types()
if error:
return ResponseBuilder.error(error)
return ResponseBuilder.success({
"resourceTypes": resource_types,
"count": len(resource_types)
})
@cluster_bp.route('/api/cluster/generate', methods=['POST'])
@@ -94,7 +85,7 @@ def generate_cluster_diagram():
return ResponseBuilder.validation_error("outputFormat", error_msg)
# Extra arguments validation
is_valid, error_msg = InputValidator.validate_extra_args(extra_args)
is_valid, error_msg = InputValidator.validate_extra_args(extra_args, "kubectl-diagrams")
if not is_valid:
return ResponseBuilder.validation_error("extraArgs", error_msg)
webapp/backend/routes/helm.py
+1 -1
View File
@@ -32,7 +32,7 @@ def generate_helm_diagram():
return ResponseBuilder.validation_error("outputFormat", error_msg)
# Extra arguments validation
is_valid, error_msg = InputValidator.validate_extra_args(extra_args)
is_valid, error_msg = InputValidator.validate_extra_args(extra_args, "helm-diagrams")
if not is_valid:
return ResponseBuilder.validation_error("extraArgs", error_msg)
webapp/backend/routes/helmfile.py
+1 -1
View File
@@ -40,7 +40,7 @@ def generate_helmfile_diagram():
return ResponseBuilder.validation_error("outputFormat", error_msg)
# Extra arguments validation
is_valid, error_msg = InputValidator.validate_extra_args(extra_args)
is_valid, error_msg = InputValidator.validate_extra_args(extra_args, "kube-diagrams")
if not is_valid:
return ResponseBuilder.validation_error("extraArgs", error_msg)
webapp/backend/routes/manifest.py
+1 -1
View File
@@ -44,7 +44,7 @@ def generate_diagram():
return ResponseBuilder.validation_error("outputFormat", error_msg)
# Extra arguments validation
is_valid, error_msg = InputValidator.validate_extra_args(extra_args)
is_valid, error_msg = InputValidator.validate_extra_args(extra_args, "kube-diagrams")
if not is_valid:
return ResponseBuilder.validation_error("extraArgs", error_msg)
webapp/backend/services/clusterService.py
+64 -63
View File
@@ -7,9 +7,12 @@ import tempfile
from typing import List, Optional, Dict, Any
from constants import MIME_TYPES
from utils import get_app_logger, log_unexpected_error
from .models import DiagramResult
from .file_manager import FileManager
from .utils import parse_extra_args, has_fatal_error, encode_content, dot_to_dot_json
from .utils import parse_extra_args, has_fatal_error, encode_content, dot_to_dot_json, redact_temp_paths
logger = get_app_logger(__name__)
COMMON_RESOURCE_TYPES = frozenset({
'pods', 'services', 'deployments', 'replicasets', 'statefulsets',
@@ -33,68 +36,69 @@ _CANNOT_REACH_API = (
)
def _raise_connection_error(error_msg: str, fallback_msg: str) -> None:
"""Raise RuntimeError with a user-friendly message based on kubectl's connection error output."""
def _connection_error_message(error_msg: str, fallback_msg: str) -> str:
"""Return a user-friendly message based on kubectl's connection error output."""
if "connect: no route to host" in error_msg or "dial tcp" in error_msg:
raise RuntimeError(
return (
"Cannot reach the Kubernetes API server (no route to host). "
"Start your cluster (e.g. minikube start, kind create cluster) "
"and verify your kubeconfig with: kubectl config current-context"
)
if "Unable to connect to the server" in error_msg:
raise RuntimeError(_CANNOT_REACH_API)
return _CANNOT_REACH_API
if "connection refused" in error_msg.lower():
raise RuntimeError(
return (
"Connection to the Kubernetes API server was refused. "
"Make sure your cluster is running and the API server is accessible."
)
raise RuntimeError(fallback_msg)
return fallback_msg
def _run_kubectl(cmd: List[str], timeout: int) -> subprocess.CompletedProcess:
"""
Run a kubectl command and return the completed process.
Raises RuntimeError if kubectl is not installed or the command times out.
CalledProcessError propagates for the caller to handle context-specifically.
"""
def _run_kubectl(cmd: List[str], timeout: int) -> tuple[Optional[subprocess.CompletedProcess], Optional[str]]:
try:
return subprocess.run(cmd, check=True, capture_output=True, text=True, timeout=timeout)
return subprocess.run(cmd, check=False, capture_output=True, text=True, timeout=timeout), None
except FileNotFoundError:
raise RuntimeError(_KUBECTL_NOT_FOUND)
return None, _KUBECTL_NOT_FOUND
except subprocess.TimeoutExpired:
raise RuntimeError(
return None, (
f"kubectl timed out after {timeout}s. "
"Check that your cluster is running and reachable, then try again."
)
def get_namespaces() -> List[str]:
def get_namespaces() -> tuple[Optional[List[str]], Optional[str]]:
"""Retrieve the sorted list of namespace names from the connected Kubernetes cluster via kubectl."""
try:
proc = _run_kubectl(["kubectl", "get", "namespaces", "-o", "json"], timeout=20)
proc, error = _run_kubectl(["kubectl", "get", "namespaces", "-o", "json"], timeout=20)
if error:
return None, error
if proc.returncode != 0:
error_msg = proc.stderr.strip() if proc.stderr else f"kubectl exited with code {proc.returncode}"
return None, _connection_error_message(error_msg, f"kubectl error while fetching namespaces: {error_msg[:200]}")
result = json.loads(proc.stdout)
return sorted([item["metadata"]["name"] for item in result.get("items", [])])
except subprocess.CalledProcessError as e:
error_msg = e.stderr.strip() if e.stderr else str(e)
_raise_connection_error(error_msg, f"kubectl error while fetching namespaces: {error_msg[:200]}")
except RuntimeError:
raise
except json.JSONDecodeError as e:
raise RuntimeError(f"Could not parse kubectl output: {str(e)}")
except Exception as e:
raise RuntimeError(f"Unexpected error while fetching namespaces: {str(e)}")
return sorted([item["metadata"]["name"] for item in result.get("items", [])]), None
except json.JSONDecodeError:
return None, log_unexpected_error(logger, "parsing kubectl output")
except Exception:
return None, log_unexpected_error(logger, "fetching namespaces")
def get_resource_types() -> List[Dict[str, Any]]:
def get_resource_types() -> tuple[Optional[List[Dict[str, Any]]], Optional[str]]:
"""
Retrieve all resource types known by the cluster via kubectl api-resources.
Each entry includes name, shortNames, namespaced scope flag, and isCommon flag.
Common types exclude events and endpoints to avoid noisy intermediate resources.
"""
try:
proc = _run_kubectl(
proc, error = _run_kubectl(
["kubectl", "api-resources", "--verbs=list", "--no-headers"], timeout=30
)
if error:
return None, error
if proc.returncode != 0:
error_msg = proc.stderr.strip() if proc.stderr else f"kubectl exited with code {proc.returncode}"
return None, _connection_error_message(error_msg, f"kubectl error while fetching resource types: {error_msg[:200]}")
resources = []
seen = set()
@@ -123,46 +127,42 @@ def get_resource_types() -> List[Dict[str, Any]]:
})
resources.sort(key=lambda x: (not x['isCommon'], x['name']))
return resources
except subprocess.CalledProcessError as e:
error_msg = e.stderr.strip() if e.stderr else str(e)
_raise_connection_error(error_msg, f"kubectl error while fetching resource types: {error_msg[:200]}")
except RuntimeError:
raise
except Exception as e:
raise RuntimeError(f"Unexpected error while fetching resource types: {str(e)}")
return resources, None
except Exception:
return None, log_unexpected_error(logger, "fetching resource types")
def get_current_context() -> str:
def get_current_context() -> tuple[Optional[str], Optional[str]]:
"""Return the name of the currently active kubectl context."""
try:
proc = _run_kubectl(["kubectl", "config", "current-context"], timeout=5)
return proc.stdout.strip()
except subprocess.CalledProcessError as e:
error_msg = e.stderr.strip() if e.stderr else str(e)
raise RuntimeError(f"No active kubectl context found: {error_msg}")
except RuntimeError:
raise
except Exception as e:
raise RuntimeError(f"Unexpected error while fetching context: {str(e)}")
proc, error = _run_kubectl(["kubectl", "config", "current-context"], timeout=5)
if error:
return None, error
if proc.returncode != 0:
error_msg = proc.stderr.strip() if proc.stderr else f"kubectl exited with code {proc.returncode}"
return None, f"No active kubectl context found: {error_msg}"
return proc.stdout.strip(), None
except Exception:
return None, log_unexpected_error(logger, "fetching context")
def _make_diagrams_error(stdout: str, stderr: str, cmd: List[str]) -> DiagramResult:
def _make_diagrams_error(stdout: str, stderr: str, cmd: List[str], *paths: str) -> DiagramResult:
"""Return a DiagramResult describing why kubectl-diagrams failed."""
command = redact_temp_paths(" ".join(cmd), *paths)
if "Unable to connect" in stderr or "connect: no route to host" in stderr:
return DiagramResult(
success=False,
error="Cannot reach the Kubernetes API server. "
"Start your cluster (e.g. minikube start, kind create cluster) "
"and verify your kubeconfig with: kubectl config current-context",
command=" ".join(cmd),
command=command,
stdout=stdout,
stderr=stderr,
)
return DiagramResult(
success=False,
error="kubectl-diagrams failed. See command output below.",
command=" ".join(cmd),
command=command,
stdout=stdout,
stderr=stderr,
)
@@ -178,6 +178,7 @@ def generate_from_cluster(
) -> DiagramResult:
"""Generate diagram using kubectl-diagrams plugin directly."""
cmd: List[str] = []
requested_output = png_output = dot_output = None
try:
resources_arg = ','.join(resource_types)
base_name = f"cluster-diagram-{uuid.uuid4().hex[:8]}"
@@ -201,21 +202,21 @@ def generate_from_cluster(
cmd.append("--without-namespace")
if extra_args.strip():
cmd.extend(parse_extra_args(extra_args))
cmd.extend(parse_extra_args(extra_args, "kubectl-diagrams"))
proc = subprocess.run(cmd, check=False, capture_output=True, text=True, timeout=60)
stdout_output = proc.stdout or ""
stderr_output = proc.stderr or ""
if proc.returncode != 0 or has_fatal_error(stdout_output, stderr_output):
return _make_diagrams_error(stdout_output, stderr_output, cmd)
return _make_diagrams_error(stdout_output, stderr_output, cmd, requested_output, png_output, dot_output)
if output_format == "dot_json":
if not os.path.exists(dot_output):
return DiagramResult(
success=False,
error=f"Output file not found: {dot_output}",
command=" ".join(cmd),
error=f"Output file not found: {os.path.basename(dot_output)}",
command=redact_temp_paths(" ".join(cmd), requested_output, png_output, dot_output),
stdout=stdout_output,
stderr=stderr_output
)
@@ -224,7 +225,7 @@ def generate_from_cluster(
return DiagramResult(
success=False,
error="dot -Tjson conversion failed (is graphviz installed?).",
command=" ".join(cmd),
command=redact_temp_paths(" ".join(cmd), requested_output, png_output, dot_output),
stdout=stdout_output,
stderr=stderr_output
)
@@ -234,8 +235,8 @@ def generate_from_cluster(
if output_info is None:
return DiagramResult(
success=False,
error=f"Output file not found: {requested_output}",
command=" ".join(cmd),
error=f"Output file not found: {os.path.basename(requested_output)}",
command=redact_temp_paths(" ".join(cmd), requested_output, png_output, dot_output),
stdout=stdout_output,
stderr=stderr_output
)
@@ -251,7 +252,7 @@ def generate_from_cluster(
mime_type=MIME_TYPES.get(produced_format, "application/octet-stream"),
filename=f"{base_name}.{produced_format}",
message="Diagram successfully generated from cluster resources using kubectl-diagrams.",
command=" ".join(cmd),
command=redact_temp_paths(" ".join(cmd), requested_output, png_output, dot_output),
stdout=stdout_output,
stderr=stderr_output
)
@@ -268,11 +269,11 @@ def generate_from_cluster(
return DiagramResult(
success=False,
error="Command timed out. The cluster might be slow or unresponsive.",
command=" ".join(cmd) or "kubectl-diagrams"
command=redact_temp_paths(" ".join(cmd), requested_output, png_output, dot_output) or "kubectl-diagrams"
)
except Exception as e:
except Exception:
return DiagramResult(
success=False,
error=f"Unexpected error: {str(e)}",
command=" ".join(cmd) or "kubectl-diagrams"
error=log_unexpected_error(logger, "generating diagram from cluster"),
command=redact_temp_paths(" ".join(cmd), requested_output, png_output, dot_output) or "kubectl-diagrams"
)
webapp/backend/services/file_manager.py
+8 -1
View File
@@ -4,6 +4,8 @@ import tempfile
from typing import Optional
from contextlib import contextmanager
from .utils import get_safe_format_extension
class FileManager:
"""Temporary file manager."""
@staticmethod
@@ -57,9 +59,14 @@ class FileManager:
output_format: Requested output format
Returns:
tuple: (requested format_path, fallback png_path)
Raises:
ValueError: If output_format is not a supported/known format.
"""
safe_ext = get_safe_format_extension(output_format)
base_without_ext = os.path.splitext(base_path)[0]
requested_output = f"{base_without_ext}.{output_format}"
requested_output = f"{base_without_ext}.{safe_ext}"
png_output = f"{base_without_ext}.png"
return requested_output, png_output
webapp/backend/services/helmService.py
+34 -25
View File
@@ -1,12 +1,17 @@
"""Service for generating diagrams from Helm charts."""
import subprocess
import os
import tempfile
import uuid
from urllib.parse import urlparse
from constants import MIME_TYPES
from utils import InputValidator, get_app_logger, log_unexpected_error
from .models import DiagramResult
from .file_manager import FileManager
from .utils import parse_extra_args, has_fatal_error, encode_content, dot_to_dot_json
from .utils import parse_extra_args, has_fatal_error, encode_content, dot_to_dot_json, get_safe_format_extension, redact_temp_paths
logger = get_app_logger(__name__)
def generate_from_helm(
@@ -25,23 +30,34 @@ def generate_from_helm(
Returns:
DiagramResult: Result of the generation
"""
# Extract base name for output file
safe_ext = get_safe_format_extension(output_format)
if not InputValidator.is_valid_helm_url(chart_url):
raise ValueError("Invalid Helm chart URL format.")
# Friendly display name derived from the chart, sanitized. Used only for
# the filename metadata returned to the client, never for a real path.
parsed = urlparse(chart_url)
base_name = os.path.basename(parsed.path).replace(".tgz", "").replace(".tar.gz", "")
display_name = os.path.basename(parsed.path).replace(".tgz", "").replace(".tar.gz", "")
# OCI URLs use the last path segment as chart name
if chart_url.startswith('oci://'):
base_name = chart_url.rstrip('/').split('/')[-1]
display_name = chart_url.rstrip('/').split('/')[-1]
dot_output = os.path.abspath(f"{base_name}.dot") if output_format == "dot_json" else None
requested_output = os.path.abspath(f"{base_name}.{output_format}")
png_output = os.path.abspath(f"{base_name}.png")
display_name = InputValidator.sanitize_filename(display_name) or "chart"
# Actual server-side output path: fully random, no link to user input.
output_base = os.path.join(tempfile.gettempdir(), f"helm-diagram-{uuid.uuid4().hex}")
dot_output = f"{output_base}.dot" if output_format == "dot_json" else None
requested_output = f"{output_base}.{safe_ext}"
png_output = f"{output_base}.png"
try:
# Command uses helm-diagrams instead of helm
cmd = ["helm-diagrams", chart_url, "-o", dot_output or requested_output]
if extra_args.strip():
cmd.extend(parse_extra_args(extra_args))
cmd.extend(parse_extra_args(extra_args, "helm-diagrams"))
# Run the command and capture output
proc = subprocess.run(cmd, check=False, capture_output=True, text=True)
@@ -79,7 +95,7 @@ def generate_from_helm(
return DiagramResult(
success=False,
error=main_error,
command=" ".join(cmd),
command=redact_temp_paths(" ".join(cmd), requested_output, png_output, dot_output),
stdout=stdout_output,
stderr=stderr_output
)
@@ -89,8 +105,8 @@ def generate_from_helm(
if not os.path.exists(dot_output):
return DiagramResult(
success=False,
error=f"Output file not found: {dot_output}",
command=" ".join(cmd),
error=f"Output file not found: {os.path.basename(dot_output)}",
command=redact_temp_paths(" ".join(cmd), requested_output, png_output, dot_output),
stdout=stdout_output,
stderr=stderr_output
)
@@ -99,7 +115,7 @@ def generate_from_helm(
return DiagramResult(
success=False,
error="dot -Tjson conversion failed (is graphviz installed?).",
command=" ".join(cmd),
command=redact_temp_paths(" ".join(cmd), requested_output, png_output, dot_output),
stdout=stdout_output,
stderr=stderr_output
)
@@ -111,7 +127,7 @@ def generate_from_helm(
return DiagramResult(
success=False,
error=f"Output file not found (looked for {os.path.basename(requested_output)} and {os.path.basename(png_output)}).",
command=" ".join(cmd),
command=redact_temp_paths(" ".join(cmd), requested_output, png_output, dot_output),
stdout=stdout_output,
stderr=stderr_output
)
@@ -131,25 +147,18 @@ def generate_from_helm(
success=True,
diagram=encoded,
mime_type=MIME_TYPES.get(produced_format, "application/octet-stream"),
filename=f"{base_name}.{produced_format}",
filename=f"{display_name}.{produced_format}",
message=message.strip(),
command=" ".join(cmd),
command=redact_temp_paths(" ".join(cmd), requested_output, png_output, dot_output),
stdout=stdout_output,
stderr=stderr_output
)
except ValueError as e:
except Exception:
FileManager.cleanup_files(requested_output, png_output, dot_output)
return DiagramResult(
success=False,
error=str(e),
command=" ".join(cmd) if 'cmd' in locals() else None
)
except Exception as e:
FileManager.cleanup_files(requested_output, png_output, dot_output)
return DiagramResult(
success=False,
error=f"Internal error: {e}",
command=" ".join(cmd) if 'cmd' in locals() else None
error=log_unexpected_error(logger, "generating diagram from Helm chart"),
command=redact_temp_paths(" ".join(cmd), requested_output, png_output, dot_output) if 'cmd' in locals() else None
)
webapp/backend/services/helmfileService.py
+19 -21
View File
@@ -3,9 +3,12 @@ import subprocess
import os
from constants import MIME_TYPES
from utils import get_app_logger, log_unexpected_error
from .models import DiagramResult
from .file_manager import FileManager
from .utils import parse_extra_args, has_fatal_error, encode_content, dot_to_dot_json
from .utils import parse_extra_args, has_fatal_error, encode_content, dot_to_dot_json, get_safe_format_extension, redact_temp_paths
logger = get_app_logger(__name__)
def generate_from_helmfile(
@@ -26,8 +29,10 @@ def generate_from_helmfile(
Returns:
DiagramResult: Result of the generation
"""
safe_ext = get_safe_format_extension(output_format)
with FileManager.create_temp_file(helmfile_content, suffix=".yaml", mode='wb') as temp_helmfile_path:
output_path = temp_helmfile_path + f".{output_format}"
output_path = temp_helmfile_path + f".{safe_ext}"
dot_output_path = temp_helmfile_path + ".dot" if output_format == "dot_json" else None
try:
@@ -47,7 +52,7 @@ def generate_from_helmfile(
return DiagramResult(
success=False,
error="Helmfile template failed. See command output below.",
command=" ".join(template_cmd),
command=redact_temp_paths(" ".join(template_cmd), temp_helmfile_path),
stdout="",
stderr=helm_err or ""
)
@@ -57,7 +62,7 @@ def generate_from_helmfile(
if without_namespace:
cmd.append("--without-namespace")
if extra_args.strip():
cmd.extend(parse_extra_args(extra_args))
cmd.extend(parse_extra_args(extra_args, "kube-diagrams"))
kube_proc = subprocess.run(
cmd,
@@ -75,7 +80,7 @@ def generate_from_helmfile(
return DiagramResult(
success=False,
error="kube-diagrams failed",
command=f"{' '.join(template_cmd)} | {' '.join(cmd)}",
command=redact_temp_paths(f"{' '.join(template_cmd)} | {' '.join(cmd)}", temp_helmfile_path, output_path, dot_output_path),
stdout=stdout_output,
stderr=stderr_output
)
@@ -84,8 +89,8 @@ def generate_from_helmfile(
if not os.path.exists(dot_output_path):
return DiagramResult(
success=False,
error=f"Output file not found: {dot_output_path}",
command=f"{' '.join(template_cmd)} | {' '.join(cmd)}",
error=f"Output file not found: {os.path.basename(dot_output_path)}",
command=redact_temp_paths(f"{' '.join(template_cmd)} | {' '.join(cmd)}", temp_helmfile_path, output_path, dot_output_path),
stdout=stdout_output,
stderr=stderr_output
)
@@ -94,15 +99,15 @@ def generate_from_helmfile(
return DiagramResult(
success=False,
error="dot -Tjson conversion failed (is graphviz installed?).",
command=f"{' '.join(template_cmd)} | {' '.join(cmd)}",
command=redact_temp_paths(f"{' '.join(template_cmd)} | {' '.join(cmd)}", temp_helmfile_path, output_path, dot_output_path),
stdout=stdout_output,
stderr=stderr_output
)
elif not os.path.exists(output_path):
return DiagramResult(
success=False,
error=f"Output file not found: {output_path}",
command=f"{' '.join(template_cmd)} | {' '.join(cmd)}",
error=f"Output file not found: {os.path.basename(output_path)}",
command=redact_temp_paths(f"{' '.join(template_cmd)} | {' '.join(cmd)}", temp_helmfile_path, output_path, dot_output_path),
stdout=stdout_output,
stderr=stderr_output
)
@@ -119,22 +124,15 @@ def generate_from_helmfile(
mime_type=MIME_TYPES.get(output_format, "application/octet-stream"),
filename=f"helmfile-diagram.{output_format}",
message="Helmfile diagram successfully generated.",
command=f"{' '.join(template_cmd)} | {' '.join(cmd)}",
command=redact_temp_paths(f"{' '.join(template_cmd)} | {' '.join(cmd)}", temp_helmfile_path, output_path, dot_output_path),
stdout=stdout_output,
stderr=stderr_output
)
except ValueError as e:
except Exception:
FileManager.cleanup_files(output_path, dot_output_path)
return DiagramResult(
success=False,
error=str(e),
command=" ".join(cmd) if 'cmd' in locals() else None
)
except Exception as e:
FileManager.cleanup_files(output_path, dot_output_path)
return DiagramResult(
success=False,
error=str(e),
command=" ".join(cmd) if 'cmd' in locals() else None
error=log_unexpected_error(logger, "generating diagram from Helmfile"),
command=redact_temp_paths(" ".join(cmd), temp_helmfile_path, output_path, dot_output_path) if 'cmd' in locals() else None
)
webapp/backend/services/manifestService.py
+15 -19
View File
@@ -3,9 +3,12 @@ import os
import subprocess
from constants import MIME_TYPES
from utils import get_app_logger, log_unexpected_error
from .models import DiagramResult
from .file_manager import FileManager
from .utils import parse_extra_args, has_fatal_error, encode_content, dot_to_dot_json
from .utils import parse_extra_args, has_fatal_error, encode_content, dot_to_dot_json, redact_temp_paths
logger = get_app_logger(__name__)
def generate_from_manifest(
@@ -39,7 +42,7 @@ def generate_from_manifest(
if without_namespace:
cmd.append("--without-namespace")
if extra_args.strip():
cmd.extend(parse_extra_args(extra_args))
cmd.extend(parse_extra_args(extra_args, "kube-diagrams"))
# Execution
proc = subprocess.run(cmd, check=False, capture_output=True, text=True)
@@ -52,7 +55,7 @@ def generate_from_manifest(
return DiagramResult(
success=False,
error="KubeDiagrams failed. See command output below.",
command=" ".join(cmd),
command=redact_temp_paths(" ".join(cmd), tmp_manifest, requested_output, png_output, dot_output),
stdout=stdout_output,
stderr=stderr_output
)
@@ -61,8 +64,8 @@ def generate_from_manifest(
if not os.path.exists(dot_output):
return DiagramResult(
success=False,
error=f"Output file not found: {dot_output}",
command=" ".join(cmd),
error=f"Output file not found: {os.path.basename(dot_output)}",
command=redact_temp_paths(" ".join(cmd), tmp_manifest, requested_output, png_output, dot_output),
stdout=stdout_output,
stderr=stderr_output
)
@@ -71,7 +74,7 @@ def generate_from_manifest(
return DiagramResult(
success=False,
error="dot -Tjson conversion failed (is graphviz installed?).",
command=" ".join(cmd),
command=redact_temp_paths(" ".join(cmd), tmp_manifest, requested_output, png_output, dot_output),
stdout=stdout_output,
stderr=stderr_output
)
@@ -82,8 +85,8 @@ def generate_from_manifest(
if not output_info:
return DiagramResult(
success=False,
error=f"Output file not found (looked for {requested_output} and {png_output}).",
command=" ".join(cmd),
error=f"Output file not found (looked for {os.path.basename(requested_output)} and {os.path.basename(png_output)}).",
command=redact_temp_paths(" ".join(cmd), tmp_manifest, requested_output, png_output, dot_output),
stdout=stdout_output,
stderr=stderr_output
)
@@ -101,23 +104,16 @@ def generate_from_manifest(
mime_type=MIME_TYPES.get(produced_format, "application/octet-stream"),
filename=f"{base_name}.{produced_format}",
message="Diagram successfully generated.",
command=" ".join(cmd),
command=redact_temp_paths(" ".join(cmd), tmp_manifest, requested_output, png_output, dot_output),
stdout=stdout_output,
stderr=stderr_output
)
except ValueError as e:
except Exception:
FileManager.cleanup_files(requested_output, png_output, dot_output)
return DiagramResult(
success=False,
error=str(e),
command=" ".join(cmd) if 'cmd' in locals() else None
)
except Exception as e:
FileManager.cleanup_files(requested_output, png_output, dot_output)
return DiagramResult(
success=False,
error=f"Internal error: {e}",
command=" ".join(cmd) if 'cmd' in locals() else None
error=log_unexpected_error(logger, "generating diagram from manifest"),
command=redact_temp_paths(" ".join(cmd), tmp_manifest, requested_output, png_output, dot_output) if 'cmd' in locals() else None
)
webapp/backend/services/utils.py
+47 -7
View File
@@ -6,7 +6,15 @@ import re
import shlex
import subprocess
from constants import TEXT_FORMATS
from constants import TEXT_FORMATS, MIME_TYPES
from utils import InputValidator
def redact_temp_paths(text: str, *paths: str) -> str:
for path in paths:
if path:
text = text.replace(path, os.path.basename(path))
return text
def has_fatal_error(stdout_txt: str, stderr_txt: str) -> bool:
"""
@@ -22,25 +30,57 @@ def has_fatal_error(stdout_txt: str, stderr_txt: str) -> bool:
return ("error:" in (stdout_txt or "").lower()) or ("error:" in (stderr_txt or "").lower())
def parse_extra_args(extra_args: str) -> list[str]:
_SAFE_FORMAT_EXTENSIONS = {fmt: fmt for fmt in MIME_TYPES}
def get_safe_format_extension(output_format: str) -> str:
"""
Parse a string of extra CLI arguments using shell-like tokenization.
Look up output_format in a hardcoded allowlist and return the matching
literal extension string.
Args:
output_format: Output format to check
Returns:
str: The same format string, sourced from a hardcoded mapping
Raises:
ValueError: If output_format is not in MIME_TYPES
"""
try:
return _SAFE_FORMAT_EXTENSIONS[output_format]
except KeyError:
raise ValueError(f"Unsupported output format: {output_format!r}")
def parse_extra_args(extra_args: str, tool: str) -> list[str]:
"""
Parse a string of extra CLI arguments using shell-like tokenization,
rejecting any flag not in that tool's allowlist (EXTRA_ARGS_ALLOWED_FLAGS).
Args:
extra_args: Space-separated argument string (may include quoted tokens)
tool: Key into EXTRA_ARGS_ALLOWED_FLAGS identifying the target CLI tool
Returns:
list[str]: Parsed argument tokens
Raises:
ValueError: If the argument string has invalid shell syntax
ValueError: If the argument string has invalid shell syntax, or
contains a flag not allowed for this tool
"""
if not extra_args or not extra_args.strip():
return []
try:
return shlex.split(extra_args.strip())
except Exception as e:
raise ValueError(f"Invalid extraArgs: {e}")
tokens = shlex.split(extra_args.strip())
except Exception:
raise ValueError("Invalid extraArgs: could not parse the value (check for unmatched quotes).")
bad_flag = InputValidator.find_disallowed_flag(tokens, tool)
if bad_flag:
raise ValueError(f"Extra arg flag '{bad_flag}' is not allowed.")
return tokens
def encode_content(content: bytes, output_format: str) -> str:
webapp/backend/utils/__init__.py
+2 -1
View File
@@ -1,5 +1,5 @@
"""Package utils."""
from .logger import AppLogger, get_app_logger
from .logger import AppLogger, get_app_logger, log_unexpected_error
from .validators import InputValidator, ValidationError
from .response_builder import ResponseBuilder
from .access_logger import get_real_ip, log_request, log_request_compact
@@ -7,6 +7,7 @@ from .access_logger import get_real_ip, log_request, log_request_compact
__all__ = [
'AppLogger',
'get_app_logger',
'log_unexpected_error',
'InputValidator',
'ValidationError',
'ResponseBuilder',
webapp/backend/utils/logger.py
+19
View File
@@ -105,3 +105,22 @@ def get_app_logger(name: str = None) -> logging.Logger:
name = frame.f_globals.get('__name__', 'app')
return AppLogger.get_logger(name)
def log_unexpected_error(logger: logging.Logger, action: str) -> str:
"""
Log the full exception for an unexpected error and return a generic,
client-safe message describing it (no internal exception detail).
Must be called from an except block (uses logger.exception, which
requires an active exception context to capture the traceback).
Args:
logger: Logger to record the full exception on
action: Short description of what was being done when it occurred
Returns:
str: Generic message safe to return to the client
"""
logger.exception(f"Unexpected error while {action}")
return f"Unexpected error while {action}. Check server logs for details."
webapp/backend/utils/validators.py
+49 -14
View File
@@ -1,7 +1,8 @@
"""Validation of user inputs."""
import re
import shlex
from typing import Optional, Tuple
from constants import MANIFEST_RE, KIND_RE
from constants import MANIFEST_RE, KIND_RE, EXTRA_ARGS_ALLOWED_FLAGS
class ValidationError(Exception):
@@ -14,10 +15,15 @@ class InputValidator:
SUPPORTED_FORMATS = ['png', 'jpg', 'jpeg', 'gif', 'svg', 'pdf', 'dot', 'dot_json', 'drawio', 'mermaid', 'd2']
# Valid Pattern for url
HELM_URL_PATTERN = re.compile(
r'^(https?://|oci://|file://|[a-zA-Z0-9\-_]+/[a-zA-Z0-9\-_]+)'
)
HELM_SCHEME_URL_PATTERN = re.compile(r'^(https?|oci|file)://[a-zA-Z0-9\-_./:~]*$')
HELM_CHART_REF_PATTERN = re.compile(r'^[a-zA-Z0-9][a-zA-Z0-9\-_]*/[a-zA-Z0-9\-_]+$')
@classmethod
def is_valid_helm_url(cls, url: str) -> bool:
return bool(
cls.HELM_SCHEME_URL_PATTERN.fullmatch(url)
or cls.HELM_CHART_REF_PATTERN.fullmatch(url)
)
@classmethod
def validate_manifest(cls, content: str) -> Tuple[bool, Optional[str]]:
@@ -83,7 +89,7 @@ class InputValidator:
url = url.strip()
# Verify the helm url pattern
if not cls.HELM_URL_PATTERN.match(url):
if not cls.is_valid_helm_url(url):
return False, "Invalid Helm chart URL format. Must start with http://, https://, oci://, file://, or be a chart reference."
return True, None
@@ -109,13 +115,35 @@ class InputValidator:
return True, None
@classmethod
def validate_extra_args(cls, args: str) -> Tuple[bool, Optional[str]]:
@staticmethod
def find_disallowed_flag(tokens: list, tool: str) -> Optional[str]:
"""
Validate extra args.
Return the first token that looks like a CLI flag not in that tool's
allowlist (EXTRA_ARGS_ALLOWED_FLAGS), or None if all tokens are allowed.
Args:
tokens: Already-tokenized extra args (see shlex.split)
tool: Key into EXTRA_ARGS_ALLOWED_FLAGS identifying the target CLI tool
Returns:
Optional[str]: The disallowed flag, or None if all tokens are allowed
"""
allowed_flags = EXTRA_ARGS_ALLOWED_FLAGS[tool]
for token in tokens:
if token.startswith('-'):
flag = token.split('=', 1)[0]
if flag not in allowed_flags:
return flag
return None
@classmethod
def validate_extra_args(cls, args: str, tool: str) -> Tuple[bool, Optional[str]]:
"""
Validate extra args against the allowlist of flags for the given tool.
Args:
args: extra_args
tool: Key into EXTRA_ARGS_ALLOWED_FLAGS identifying the target CLI tool
Returns:
Tuple[bool, Optional[str]]: (is_valid, error_message)
@@ -123,10 +151,18 @@ class InputValidator:
if not args or not args.strip():
return True, None
dangerous_chars = [';', '&', '|', '`', '$', '(', ')']
for char in dangerous_chars:
if char in args:
return False, f"Extra args contain dangerous character '{char}'"
try:
tokens = shlex.split(args.strip())
except ValueError:
return False, "Invalid extraArgs: could not parse the value (check for unmatched quotes)."
bad_flag = cls.find_disallowed_flag(tokens, tool)
if bad_flag:
allowed_flags = EXTRA_ARGS_ALLOWED_FLAGS[tool]
return False, (
f"Extra arg flag '{bad_flag}' is not allowed. "
f"Allowed flags: {', '.join(sorted(allowed_flags))}"
)
return True, None
@@ -181,7 +217,6 @@ class InputValidator:
helmfile_keys = ["\nreleases:", "\nrepositories:", "\nhelmdefaults:", "\nenvironments:", "\ntemplates:"]
return any(key in t for key in helmfile_keys)
# TODO: Not used yet
@classmethod
def sanitize_filename(cls, filename: str) -> str:
"""
webapp/frontend/src/components/common/DiagramViewer.jsx
+1 -2
View File
@@ -52,7 +52,7 @@ function DrawioViewer({ content }) {
// Listen for draw.io init event and send the XML content
useEffect(() => {
const handleMessage = (event) => {
if (!event.origin.includes('diagrams.net')) return;
if (event.origin !== 'https://embed.diagrams.net') return;
try {
const data = JSON.parse(event.data);
if (data.event === 'init') {
@@ -93,7 +93,6 @@ function DrawioViewer({ content }) {
);
}
function DiagramRenderError({ formatLabel, message }) {
return (
<div className="flex items-center justify-center w-full h-48 text-red-500 text-sm p-4 text-center">
webapp/frontend/src/hooks/useFileUpload.js
+1 -1
View File
@@ -28,7 +28,7 @@ export function useFileUpload() {
const fileName = file.name.toLowerCase();
const validExtensions = FILE_INPUT.ACCEPT.split(',').map((ext) => ext.trim());
const hasValidExtension = validExtensions.some((ext) =>
fileName.endsWith(ext.replace('*', ''))
fileName.endsWith(ext.replaceAll('*', ''))
);
if (!hasValidExtension) {
webapp/frontend/src/hooks/useHistorySync.js
+1 -1
View File
@@ -73,4 +73,4 @@ export function useHistorySync({
historyContext.clearRestoredItem();
}
}, [historyContext?.restoredItem, restoreDiagram]);
}
}
webapp/frontend/src/hooks/useSvgDiagramRenderer.js
+6 -2
View File
@@ -17,7 +17,11 @@ function fixSvgIntrinsicSize(svgEl) {
* @param {boolean} [options.showSpinner] - Whether to expose a loading state while rendering
* @returns {{ containerRef: React.RefObject, error: string|null, isRendering: boolean }}
*/
export function useSvgDiagramRenderer(renderFn, content, { formatLabel = '', showSpinner = false } = {}) {
export function useSvgDiagramRenderer(
renderFn,
content,
{ formatLabel = '', showSpinner = false } = {}
) {
const containerRef = useRef(null);
const [error, setError] = useState(null);
const [isRendering, setIsRendering] = useState(showSpinner);
@@ -47,4 +51,4 @@ export function useSvgDiagramRenderer(renderFn, content, { formatLabel = '', sho
}, [content]);
return { containerRef, error, isRendering };
}
}