574 lines
24 KiB
Python
Executable File
574 lines
24 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Validate Wild Directory app packages for common authoring errors.
|
|
|
|
Usage:
|
|
python3 admin/scripts/validate-apps.py [app-name ...]
|
|
python3 admin/scripts/validate-apps.py --list-rules
|
|
|
|
With no arguments, validates all apps. Pass app names to check specific ones.
|
|
Exit code 1 if any errors are found.
|
|
|
|
To suppress specific rules for an app, add ignoreRules to its app.yaml:
|
|
ignoreRules:
|
|
- WC-HELM
|
|
- WC-SEL
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import re
|
|
import yaml
|
|
|
|
REQUIRED_APP_YAML_FIELDS = ["name", "is", "description", "category", "latest"]
|
|
REQUIRED_MANIFEST_FIELDS = ["version"]
|
|
STANDARD_LABEL_PAIRS = {"managedBy": "kustomize", "partOf": "wild-cloud"}
|
|
HELM_LABELS = {"app.kubernetes.io/name", "app.kubernetes.io/instance", "app.kubernetes.io/component"}
|
|
MUTABLE_IMAGE_TAGS = {"latest", "main", "master", "edge", "develop", "dev", "nightly", "stable"}
|
|
GLOBAL_TEMPLATE_NAMESPACES = {"cloud", "cluster", "operator", "apps", "secrets", "app"}
|
|
|
|
|
|
def load_rules():
|
|
"""Load rule definitions from validation-rules.yaml in the same directory."""
|
|
script_dir = os.path.dirname(os.path.abspath(__file__))
|
|
rules_path = os.path.join(script_dir, "validation-rules.yaml")
|
|
with open(rules_path) as f:
|
|
data = yaml.safe_load(f)
|
|
rules = {}
|
|
remediation = {}
|
|
severity = {}
|
|
for rule_id, info in (data or {}).items():
|
|
if isinstance(info, dict):
|
|
rules[rule_id] = info.get("description", "")
|
|
severity[rule_id] = info.get("severity", "warning")
|
|
rem = info.get("remediation", "")
|
|
if rem:
|
|
remediation[rule_id] = rem.strip()
|
|
return rules, remediation, severity
|
|
|
|
|
|
RULES, REMEDIATION, SEVERITY = load_rules()
|
|
|
|
|
|
def add_error(errors, ignored_rules, rule_id, message):
|
|
if rule_id not in ignored_rules:
|
|
errors.append((rule_id, message))
|
|
|
|
|
|
def add_warning(warnings, ignored_rules, rule_id, message):
|
|
if rule_id not in ignored_rules:
|
|
warnings.append((rule_id, message))
|
|
|
|
|
|
def strip_gomplate(text):
|
|
"""Replace gomplate expressions with a placeholder so the file parses as valid YAML."""
|
|
return re.sub(r"\{\{.*?\}\}", "PLACEHOLDER", text, flags=re.DOTALL)
|
|
|
|
|
|
def load_yaml(path):
|
|
with open(path) as f:
|
|
content = f.read()
|
|
return yaml.safe_load(strip_gomplate(content))
|
|
|
|
|
|
def load_yaml_all(path):
|
|
"""Yield all YAML documents in a file (handles multi-doc files)."""
|
|
with open(path) as f:
|
|
content = f.read()
|
|
yield from yaml.safe_load_all(strip_gomplate(content))
|
|
|
|
|
|
def resource_docs(kustomize_dir, resources):
|
|
"""Yield (filepath, doc) for each YAML document in listed resource files."""
|
|
for resource in resources:
|
|
path = os.path.join(kustomize_dir, resource)
|
|
if not os.path.isfile(path):
|
|
continue
|
|
try:
|
|
for doc in load_yaml_all(path):
|
|
if doc:
|
|
yield path, doc
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def resources_have_workloads(kustomize_dir, resources):
|
|
"""Return True if any listed resource file contains a Deployment or StatefulSet."""
|
|
for _, doc in resource_docs(kustomize_dir, resources):
|
|
if doc.get("kind") in ("Deployment", "StatefulSet"):
|
|
return True
|
|
return False
|
|
|
|
|
|
def check_security_context(spec, resource_name, errors, warnings, ignored_rules):
|
|
"""Check pod-level and container-level security contexts."""
|
|
pod_spec = spec.get("template", {}).get("spec", {}) if "template" in spec else spec
|
|
if not pod_spec:
|
|
return
|
|
|
|
pod_sc = pod_spec.get("securityContext")
|
|
if not pod_sc:
|
|
add_warning(warnings, ignored_rules, "WC-SC-POD",
|
|
f"{resource_name}: no pod-level securityContext")
|
|
|
|
for container in pod_spec.get("containers", []) + pod_spec.get("initContainers", []):
|
|
cname = container.get("name", "?")
|
|
csc = container.get("securityContext")
|
|
if not csc:
|
|
add_warning(warnings, ignored_rules, "WC-SC-CTR",
|
|
f"{resource_name}/{cname}: no container-level securityContext")
|
|
|
|
|
|
def check_mysql_db_init(kustomize_dir, resources, errors, warnings, ignored_rules):
|
|
"""Check MySQL db-init SQL for the idempotent password pattern (WC-DBIN)."""
|
|
for resource in resources:
|
|
path = os.path.join(kustomize_dir, resource)
|
|
if not os.path.isfile(path):
|
|
continue
|
|
try:
|
|
with open(path) as f:
|
|
content = f.read()
|
|
except Exception:
|
|
continue
|
|
# Only files that contain MySQL-style SQL (IDENTIFIED BY is MySQL-specific)
|
|
if "CREATE USER" not in content or "IDENTIFIED BY" not in content:
|
|
continue
|
|
if "ALTER USER" not in content:
|
|
add_error(errors, ignored_rules, "WC-DBIN",
|
|
f"{os.path.basename(path)}: MySQL CREATE USER IF NOT EXISTS without ALTER USER"
|
|
" — passwords won't update on redeploy (see note #17)")
|
|
|
|
|
|
def check_docker_dns(kustomize_dir, resources, errors, warnings, ignored_rules):
|
|
"""Check for Docker DNS resolver (127.0.0.11) which doesn't work in Kubernetes (WC-DKRDNS)."""
|
|
for resource in resources:
|
|
path = os.path.join(kustomize_dir, resource)
|
|
if not os.path.isfile(path):
|
|
continue
|
|
try:
|
|
with open(path) as f:
|
|
content = f.read()
|
|
except Exception:
|
|
continue
|
|
if "127.0.0.11" in content:
|
|
add_error(errors, ignored_rules, "WC-DKRDNS",
|
|
f"{os.path.basename(path)}: contains Docker DNS resolver (127.0.0.11)"
|
|
" — use CoreDNS instead (see note #24)")
|
|
|
|
|
|
def check_resource_parse_errors(kustomize_dir, resources, errors, warnings, ignored_rules):
|
|
"""Check that all listed resource files parse as valid YAML (WC-RES-PARSE)."""
|
|
for resource in resources:
|
|
path = os.path.join(kustomize_dir, resource)
|
|
if not os.path.isfile(path):
|
|
continue # WC-RES already handles missing files
|
|
try:
|
|
list(load_yaml_all(path))
|
|
except Exception as e:
|
|
add_error(errors, ignored_rules, "WC-RES-PARSE",
|
|
f"{resource}: YAML parse error: {e}")
|
|
|
|
|
|
def check_namespace_resource(kustomize_dir, resources, errors, warnings, ignored_rules):
|
|
"""Check that a Namespace resource exists when workloads are present (WC-NSFILE)."""
|
|
if not resources_have_workloads(kustomize_dir, resources):
|
|
return
|
|
for _, doc in resource_docs(kustomize_dir, resources):
|
|
if doc.get("kind") == "Namespace":
|
|
return
|
|
add_warning(warnings, ignored_rules, "WC-NSFILE",
|
|
"no Namespace resource in kustomization.yaml — add namespace.yaml")
|
|
|
|
|
|
def check_template_vars(kustomize_dir, default_config, errors, warnings, ignored_rules):
|
|
"""Check top-level template variables in resource files against defaultConfig (WC-TVAR)."""
|
|
if not default_config:
|
|
return
|
|
defined_keys = set(default_config.keys())
|
|
for fname in sorted(os.listdir(kustomize_dir)):
|
|
if not fname.endswith(".yaml") or fname == "manifest.yaml":
|
|
continue
|
|
path = os.path.join(kustomize_dir, fname)
|
|
try:
|
|
with open(path) as f:
|
|
content = f.read()
|
|
except Exception:
|
|
continue
|
|
seen_undefined = set()
|
|
for block in re.finditer(r"\{\{-?\s*(.*?)\s*-?\}\}", content, re.DOTALL):
|
|
for m in re.finditer(r"(?<![.\w])\.([a-z]\w*)", block.group(1)):
|
|
var_name = m.group(1)
|
|
if var_name in GLOBAL_TEMPLATE_NAMESPACES:
|
|
continue
|
|
if var_name not in defined_keys and var_name not in seen_undefined:
|
|
seen_undefined.add(var_name)
|
|
add_warning(warnings, ignored_rules, "WC-TVAR",
|
|
f"{fname}: template variable .{var_name!r} not in defaultConfig")
|
|
|
|
|
|
def validate_resources(kustomize_dir, resources, errors, warnings, ignored_rules):
|
|
"""Run per-document checks on all resource files."""
|
|
rwo_pvc_names = set()
|
|
|
|
# First pass: collect RWO PVC claim names
|
|
for _, doc in resource_docs(kustomize_dir, resources):
|
|
if doc.get("kind") == "PersistentVolumeClaim":
|
|
access_modes = doc.get("spec", {}).get("accessModes", [])
|
|
if "ReadWriteOnce" in access_modes:
|
|
rwo_pvc_names.add(doc.get("metadata", {}).get("name", ""))
|
|
|
|
# Second pass: validate each document
|
|
for fpath, doc in resource_docs(kustomize_dir, resources):
|
|
fname = os.path.basename(fpath)
|
|
kind = doc.get("kind", "")
|
|
name = doc.get("metadata", {}).get("name", fname)
|
|
label_ref = f"{fname}:{name}"
|
|
|
|
if kind in ("Deployment", "StatefulSet", "DaemonSet"):
|
|
spec = doc.get("spec", {})
|
|
|
|
# Security context
|
|
check_security_context(spec, label_ref, errors, warnings, ignored_rules)
|
|
|
|
# RWO PVC → must use Recreate strategy
|
|
if kind == "Deployment":
|
|
volumes = spec.get("template", {}).get("spec", {}).get("volumes", [])
|
|
uses_rwo = any(
|
|
v.get("persistentVolumeClaim", {}).get("claimName") in rwo_pvc_names
|
|
for v in volumes
|
|
)
|
|
if uses_rwo:
|
|
strategy = spec.get("strategy", {}).get("type", "RollingUpdate")
|
|
if strategy != "Recreate":
|
|
add_error(errors, ignored_rules, "WC-RWO",
|
|
f"{label_ref}: ReadWriteOnce PVC requires strategy: Recreate (got {strategy!r})")
|
|
|
|
# No Helm-style labels in pod template
|
|
pod_labels = spec.get("template", {}).get("metadata", {}).get("labels", {}) or {}
|
|
for helm_label in HELM_LABELS:
|
|
if helm_label in pod_labels:
|
|
add_error(errors, ignored_rules, "WC-HELM",
|
|
f"{label_ref}: Helm-style label {helm_label!r} found — use simple component labels instead")
|
|
|
|
# Image tags and sources
|
|
pod_spec = spec.get("template", {}).get("spec", {}) or {}
|
|
for container in pod_spec.get("containers", []) + pod_spec.get("initContainers", []):
|
|
image = container.get("image", "")
|
|
cname = container.get("name", "?")
|
|
tag = image.split(":")[-1] if ":" in image else ""
|
|
if tag.lower() in MUTABLE_IMAGE_TAGS:
|
|
add_error(errors, ignored_rules, "WC-IMG",
|
|
f"{label_ref}/{cname}: image tag {tag!r} is mutable — use a specific version tag")
|
|
elif not tag:
|
|
add_warning(warnings, ignored_rules, "WC-IMG",
|
|
f"{label_ref}/{cname}: image has no tag — specify a version")
|
|
if re.search(r"(?:^|/)bitnami", image.split(":")[0]):
|
|
add_warning(warnings, ignored_rules, "WC-BITNAMI",
|
|
f"{label_ref}/{cname}: uses a Bitnami image ({image.split(':')[0]!r})"
|
|
" — prefer an official image")
|
|
|
|
elif kind == "Ingress":
|
|
annotations = doc.get("metadata", {}).get("annotations", {}) or {}
|
|
spec = doc.get("spec", {}) or {}
|
|
|
|
# Must use spec.ingressClassName, not annotation
|
|
if "kubernetes.io/ingress.class" in annotations:
|
|
add_error(errors, ignored_rules, "WC-ING-ANN",
|
|
f"{label_ref}: use spec.ingressClassName: traefik, not kubernetes.io/ingress.class annotation")
|
|
if spec.get("ingressClassName") != "traefik":
|
|
add_warning(warnings, ignored_rules, "WC-ING-CLS",
|
|
f"{label_ref}: spec.ingressClassName should be 'traefik' (got {spec.get('ingressClassName')!r})")
|
|
|
|
# No cert-manager annotation (wildcard cert is pre-distributed)
|
|
if "cert-manager.io/cluster-issuer" in annotations:
|
|
add_error(errors, ignored_rules, "WC-CERT",
|
|
f"{label_ref}: do not use cert-manager.io/cluster-issuer — wildcard TLS cert is pre-distributed to namespaces")
|
|
|
|
# External-dns target annotation
|
|
if "external-dns.alpha.kubernetes.io/target" not in annotations:
|
|
add_warning(warnings, ignored_rules, "WC-DNS",
|
|
f"{label_ref}: missing external-dns.alpha.kubernetes.io/target annotation")
|
|
|
|
|
|
def validate_kustomization(kustomize_dir, app_name, default_config, errors, warnings, ignored_rules):
|
|
path = os.path.join(kustomize_dir, "kustomization.yaml")
|
|
if not os.path.exists(path):
|
|
add_error(errors, ignored_rules, "WC-KUS-MISS", "MISSING kustomization.yaml")
|
|
return
|
|
|
|
try:
|
|
data = load_yaml(path)
|
|
except Exception as e:
|
|
add_error(errors, ignored_rules, "WC-KUS-PARSE", f"PARSE ERROR in kustomization.yaml: {e}")
|
|
return
|
|
|
|
if not data:
|
|
add_error(errors, ignored_rules, "WC-KUS-EMPTY", "EMPTY kustomization.yaml")
|
|
return
|
|
|
|
# Check all listed resources exist on disk
|
|
resources = data.get("resources") or []
|
|
for resource in resources:
|
|
resource_path = os.path.join(kustomize_dir, resource)
|
|
if not os.path.exists(resource_path):
|
|
add_error(errors, ignored_rules, "WC-RES", f"resource not found: {resource}")
|
|
|
|
has_workloads = resources_have_workloads(kustomize_dir, resources)
|
|
|
|
if has_workloads:
|
|
if "namespace" not in data:
|
|
add_warning(warnings, ignored_rules, "WC-NS",
|
|
"no namespace set in kustomization.yaml (has Deployment/StatefulSet)")
|
|
|
|
# Require includeSelectors: true
|
|
labels = data.get("labels") or []
|
|
include_selectors_entries = [
|
|
lbl for lbl in labels
|
|
if isinstance(lbl, dict) and lbl.get("includeSelectors") is True
|
|
]
|
|
if not include_selectors_entries:
|
|
add_error(errors, ignored_rules, "WC-SEL",
|
|
"labels.includeSelectors: true missing (required when Deployment/StatefulSet present)")
|
|
else:
|
|
pairs = include_selectors_entries[0].get("pairs") or {}
|
|
for key, expected in STANDARD_LABEL_PAIRS.items():
|
|
if pairs.get(key) != expected:
|
|
add_warning(warnings, ignored_rules, "WC-PAIRS",
|
|
f"label pair {key}: {expected!r} missing or wrong")
|
|
# app label should match the app directory name
|
|
if pairs.get("app") and pairs.get("app") != app_name:
|
|
add_warning(warnings, ignored_rules, "WC-LBL",
|
|
f"label pair app: {pairs['app']!r} doesn't match app directory name {app_name!r}")
|
|
|
|
# Per-resource checks
|
|
check_resource_parse_errors(kustomize_dir, resources, errors, warnings, ignored_rules)
|
|
validate_resources(kustomize_dir, resources, errors, warnings, ignored_rules)
|
|
check_mysql_db_init(kustomize_dir, resources, errors, warnings, ignored_rules)
|
|
check_docker_dns(kustomize_dir, resources, errors, warnings, ignored_rules)
|
|
check_namespace_resource(kustomize_dir, resources, errors, warnings, ignored_rules)
|
|
check_template_vars(kustomize_dir, default_config, errors, warnings, ignored_rules)
|
|
|
|
|
|
def validate_manifest(version_dir, errors, warnings, ignored_rules):
|
|
manifest_path = os.path.join(version_dir, "manifest.yaml")
|
|
if not os.path.exists(manifest_path):
|
|
add_error(errors, ignored_rules, "WC-MAN-MISS", "MISSING manifest.yaml")
|
|
return
|
|
|
|
try:
|
|
manifest = load_yaml(manifest_path)
|
|
except Exception as e:
|
|
add_error(errors, ignored_rules, "WC-MAN-PARSE", f"PARSE ERROR in manifest.yaml: {e}")
|
|
return
|
|
|
|
if not manifest:
|
|
add_error(errors, ignored_rules, "WC-MAN-EMPTY", "EMPTY manifest.yaml")
|
|
return
|
|
|
|
for field in REQUIRED_MANIFEST_FIELDS:
|
|
if field not in manifest:
|
|
add_error(errors, ignored_rules, "WC-MAN-REQ",
|
|
f"manifest.yaml missing field: {field}")
|
|
|
|
# defaultConfig should have a namespace field
|
|
default_config = manifest.get("defaultConfig") or {}
|
|
if not default_config:
|
|
add_warning(warnings, ignored_rules, "WC-CFG",
|
|
"manifest.yaml: defaultConfig is empty or missing")
|
|
elif "namespace" not in default_config:
|
|
add_warning(warnings, ignored_rules, "WC-CFG-NS",
|
|
"manifest.yaml: defaultConfig missing 'namespace' field")
|
|
|
|
# Check postgres dbUrl secrets contain ?sslmode=disable
|
|
for secret in manifest.get("defaultSecrets") or []:
|
|
if not isinstance(secret, dict):
|
|
continue
|
|
default = str(secret.get("default", ""))
|
|
if ("postgres" in default or "postgresql" in default) and "sslmode" not in default:
|
|
add_error(errors, ignored_rules, "WC-SSL",
|
|
f"manifest.yaml: defaultSecret {secret.get('key')!r} has postgres URL without ?sslmode=disable")
|
|
|
|
# requiredSecrets should use <app-ref>.<key> format
|
|
for rs in manifest.get("requiredSecrets") or []:
|
|
if isinstance(rs, str) and "." not in rs:
|
|
add_error(errors, ignored_rules, "WC-DOT",
|
|
f"manifest.yaml: requiredSecret {rs!r} missing dot notation (<app-ref>.<key>)")
|
|
|
|
|
|
def validate_version(version_dir, app_name, errors, warnings, ignored_rules):
|
|
validate_manifest(version_dir, errors, warnings, ignored_rules)
|
|
|
|
# Load defaultConfig for template variable checking
|
|
default_config = {}
|
|
manifest_path = os.path.join(version_dir, "manifest.yaml")
|
|
if os.path.exists(manifest_path):
|
|
try:
|
|
manifest_data = load_yaml(manifest_path) or {}
|
|
default_config = manifest_data.get("defaultConfig") or {}
|
|
except Exception:
|
|
pass
|
|
|
|
# Config-only apps (e.g. smtp) have no kubernetes resources — skip kustomization check
|
|
kustomize_path = os.path.join(version_dir, "kustomization.yaml")
|
|
resource_yamls = [
|
|
f for f in os.listdir(version_dir)
|
|
if f.endswith(".yaml") and f not in ("manifest.yaml", "kustomization.yaml")
|
|
]
|
|
if os.path.exists(kustomize_path) or resource_yamls:
|
|
validate_kustomization(version_dir, app_name, default_config, errors, warnings, ignored_rules)
|
|
|
|
|
|
def validate_app(app_dir):
|
|
errors = []
|
|
warnings = []
|
|
app_name = os.path.basename(app_dir)
|
|
|
|
app_yaml_path = os.path.join(app_dir, "app.yaml")
|
|
if not os.path.exists(app_yaml_path):
|
|
return [("WC-APP-MISS", "MISSING app.yaml")], []
|
|
|
|
try:
|
|
app_data = load_yaml(app_yaml_path)
|
|
except Exception as e:
|
|
return [("WC-APP-PARSE", f"PARSE ERROR in app.yaml: {e}")], []
|
|
|
|
if not app_data:
|
|
return [("WC-APP-EMPTY", "EMPTY app.yaml")], []
|
|
|
|
ignored_rules = set(app_data.get("ignoreRules") or [])
|
|
|
|
# Flag any ignoreRules entries that don't correspond to a known rule
|
|
for rule_id in ignored_rules:
|
|
if rule_id not in RULES:
|
|
errors.append(("WC-IGN", f"app.yaml ignoreRules contains unknown rule: {rule_id!r}"))
|
|
|
|
for field in REQUIRED_APP_YAML_FIELDS:
|
|
if not app_data.get(field):
|
|
add_error(errors, ignored_rules, "WC-REQ",
|
|
f"app.yaml missing required field: {field}")
|
|
|
|
# icon is cosmetic — warn rather than error
|
|
if not app_data.get("icon"):
|
|
add_warning(warnings, ignored_rules, "WC-ICON", "app.yaml missing field: icon")
|
|
|
|
# name must match directory name
|
|
if app_data.get("name") and app_data["name"] != app_name:
|
|
add_error(errors, ignored_rules, "WC-NAME",
|
|
f"app.yaml name {app_data['name']!r} doesn't match directory name {app_name!r}")
|
|
|
|
latest = str(app_data.get("latest", ""))
|
|
versions_dir = os.path.join(app_dir, "versions")
|
|
|
|
if not os.path.isdir(versions_dir):
|
|
add_error(errors, ignored_rules, "WC-VER", "MISSING versions/ directory")
|
|
return errors, warnings
|
|
|
|
# Check slot naming convention and validate each slot
|
|
for slot_name in sorted(os.listdir(versions_dir)):
|
|
slot_path = os.path.join(versions_dir, slot_name)
|
|
if not os.path.isdir(slot_path):
|
|
continue
|
|
# Flag full version strings (two or more dots) or packaging revision suffix (-N)
|
|
if re.search(r"\.\d+\.\d+", slot_name) or re.search(r"-\d+$", slot_name):
|
|
add_warning(warnings, ignored_rules, "WC-SLOT",
|
|
f"versions/{slot_name}: slot name looks like a full version or revision"
|
|
" — use the major version only (e.g. '1', 'v1')")
|
|
|
|
if latest:
|
|
latest_dir = os.path.join(versions_dir, latest)
|
|
if not os.path.isdir(latest_dir):
|
|
add_error(errors, ignored_rules, "WC-LATEST",
|
|
f"latest version directory not found: versions/{latest}")
|
|
else:
|
|
ver_errors, ver_warnings = [], []
|
|
validate_version(latest_dir, app_name, ver_errors, ver_warnings, ignored_rules)
|
|
errors.extend((rule_id, f"versions/{latest}: {msg}") for rule_id, msg in ver_errors)
|
|
warnings.extend((rule_id, f"versions/{latest}: {msg}") for rule_id, msg in ver_warnings)
|
|
|
|
# Check that upgrade waypoint slots exist
|
|
upgrade = app_data.get("upgrade") or {}
|
|
for rule in upgrade.get("from") or []:
|
|
via = rule.get("via")
|
|
if via:
|
|
via_dir = os.path.join(versions_dir, str(via))
|
|
if not os.path.isdir(via_dir):
|
|
add_error(errors, ignored_rules, "WC-VIA",
|
|
f"app.yaml upgrade.from[].via={via!r} references a non-existent slot")
|
|
|
|
return errors, warnings
|
|
|
|
|
|
def main():
|
|
args = sys.argv[1:]
|
|
|
|
if "--list-rules" in args:
|
|
print("Rule IDs for use in app.yaml ignoreRules:\n")
|
|
for rule_id in RULES:
|
|
sev = SEVERITY.get(rule_id, "warning")
|
|
print(f" {rule_id:<14} {sev:<8} {RULES[rule_id]}")
|
|
if rule_id in REMEDIATION:
|
|
lines = REMEDIATION[rule_id].split("\n")
|
|
print(f" {'':<14} Fix: {lines[0]}")
|
|
for line in lines[1:]:
|
|
print(f" {'':<14} {line}")
|
|
return 0
|
|
|
|
requested = [a for a in args if not a.startswith("--")]
|
|
|
|
wild_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
if requested:
|
|
app_dirs = []
|
|
for name in requested:
|
|
path = os.path.join(wild_dir, name)
|
|
if not os.path.isdir(path):
|
|
print(f"ERROR: {name} not found in wild-directory", file=sys.stderr)
|
|
sys.exit(1)
|
|
app_dirs.append(path)
|
|
else:
|
|
app_dirs = sorted(
|
|
os.path.join(wild_dir, d)
|
|
for d in os.listdir(wild_dir)
|
|
if os.path.isdir(os.path.join(wild_dir, d))
|
|
and not d.startswith(".")
|
|
and os.path.exists(os.path.join(wild_dir, d, "app.yaml"))
|
|
)
|
|
|
|
total_errors = 0
|
|
total_warnings = 0
|
|
|
|
for app_dir in app_dirs:
|
|
app_name = os.path.basename(app_dir)
|
|
errors, warnings = validate_app(app_dir)
|
|
|
|
if errors or warnings:
|
|
print(f"\n{app_name}:")
|
|
shown_remediation = set()
|
|
for rule_id, msg in errors:
|
|
print(f" ERROR [{rule_id}] {msg}")
|
|
if rule_id in REMEDIATION and rule_id not in shown_remediation:
|
|
lines = REMEDIATION[rule_id].split("\n")
|
|
print(f" Fix: {lines[0]}")
|
|
for line in lines[1:]:
|
|
print(f" {line}")
|
|
shown_remediation.add(rule_id)
|
|
for rule_id, msg in warnings:
|
|
print(f" WARN [{rule_id}] {msg}")
|
|
if rule_id in REMEDIATION and rule_id not in shown_remediation:
|
|
lines = REMEDIATION[rule_id].split("\n")
|
|
print(f" Fix: {lines[0]}")
|
|
for line in lines[1:]:
|
|
print(f" {line}")
|
|
shown_remediation.add(rule_id)
|
|
|
|
total_errors += len(errors)
|
|
total_warnings += len(warnings)
|
|
|
|
print(f"\n{'─'*50}")
|
|
print(f"checked {len(app_dirs)} apps • {total_errors} errors • {total_warnings} warnings")
|
|
|
|
return 1 if total_errors > 0 else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|