Skills › Scientific Agent Skills › dnanexus-integration › validate_dxapp.py
validate_dxapp.py
808 lines · skills/dnanexus-integration/scripts/validate_dxapp.py at 5f2bc03
1#!/usr/bin/env python3
2"""Offline validation and safety linting for a DNAnexus dxapp.json file."""
4from __future__ import annotations
6import argparse
7import json
8import re
9import sys
10from dataclasses import asdict, dataclass
11from pathlib import Path
12from typing import Any, Iterable, Optional
15NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
16APP_NAME_RE = re.compile(r"^[A-Za-z0-9._-]+$")
17# /app/new accepts freeform versions; semantic versioning is a recommendation.
18VERSION_RE = re.compile(r"^[A-Za-z0-9._+-]+$")
19SECRET_KEY_RE = re.compile(
20 r"(?:^|[_-])(token|password|passwd|secret|private[_-]?key)(?:$|[_-])",
21 re.IGNORECASE,
22)
24CLASSES = {
25 "int",
26 "float",
27 "string",
28 "boolean",
29 "hash",
30 "file",
31 "record",
32 "applet",
33 "array:int",
34 "array:float",
35 "array:string",
36 "array:boolean",
37 "array:file",
38 "array:record",
39 "array:applet",
40}
41ACCESS_LEVELS = {"VIEW", "UPLOAD", "CONTRIBUTE", "ADMINISTER"}
42INTERPRETERS = {"bash", "python3"}
43SUPPORTED_RELEASES = {"20.04", "24.04"}
44RESTARTABLE_FAILURES = {
45 "ExecutionError",
46 "UnresponsiveWorker",
47 "JMInternalError",
48 "AppInternalError",
49 "AppInsufficientResourceError",
50 "JobTimeoutExceeded",
51 "SpotInstanceInterruption",
52 "*",
53}
54PLACEHOLDER_VALUES = {
55 "",
56 "changeme",
57 "example",
58 "placeholder",
59 "redacted",
60 "<secret>",
61 "<token>",
62}
65@dataclass(frozen=True)
66class Issue:
67 severity: str
68 code: str
69 path: str
70 message: str
73class Validator:
74 def __init__(self, manifest: Any, kind: str) -> None:
75 self.manifest = manifest
76 self.requested_kind = kind
77 self.issues: list[Issue] = []
79 def add(
80 self,
81 severity: str,
82 code: str,
83 path: str,
84 message: str,
85 ) -> None:
86 self.issues.append(Issue(severity, code, path, message))
88 def error(self, code: str, path: str, message: str) -> None:
89 self.add("error", code, path, message)
91 def warning(self, code: str, path: str, message: str) -> None:
92 self.add("warning", code, path, message)
94 def validate(self) -> tuple[str, list[Issue]]:
95 if not isinstance(self.manifest, dict):
96 self.error(
97 "root-type",
98 "$",
99 "the manifest root must be a JSON object",
100 )
101 return "unknown", self.issues
103 kind = self.determine_kind()
104 self.validate_metadata(kind)
105 self.validate_specs(kind)
106 self.validate_run_spec()
107 self.validate_regional_options()
108 self.validate_access()
109 self.validate_https_app()
110 self.scan_for_embedded_secrets(self.manifest)
111 return kind, self.issues
113 def determine_kind(self) -> str:
114 if self.requested_kind != "auto":
115 return self.requested_kind
116 if "version" in self.manifest:
117 return "app"
118 return "applet"
120 def validate_metadata(self, kind: str) -> None:
121 name = self.manifest.get("name")
122 if not isinstance(name, str) or not name:
123 self.error("missing-name", "$.name", "name is required")
124 elif not APP_NAME_RE.fullmatch(name):
125 self.error(
126 "invalid-name",
127 "$.name",
128 "name may contain only letters, digits, '.', '_', and '-'",
129 )
130 elif kind == "app" and name.startswith("app-"):
131 self.error(
132 "invalid-name", "$.name", "app name cannot start with 'app-'",
133 )
135 version = self.manifest.get("version")
136 if kind == "app":
137 if not isinstance(version, str):
138 self.error(
139 "missing-version",
140 "$.version",
141 "a version string is required for apps",
142 )
143 elif not VERSION_RE.fullmatch(version):
144 self.error(
145 "invalid-version",
146 "$.version",
147 "app version must be non-empty and contain only letters, "
148 "digits, '.', '_', '+', and '-'",
149 )
150 elif version is not None and not isinstance(version, str):
151 self.error(
152 "invalid-version",
153 "$.version",
154 "version must be a string when supplied",
155 )
157 if "resources" in self.manifest:
158 self.warning(
159 "deprecated-resources",
160 "$.resources",
161 "top-level resources is deprecated; move it under "
162 "regionalOptions.<region>.resources",
163 )
165 def validate_specs(self, kind: str) -> None:
166 for spec_name in ("inputSpec", "outputSpec"):
167 value = self.manifest.get(spec_name)
168 if value is None:
169 if kind == "app":
170 self.error(
171 "missing-spec",
172 f"$.{spec_name}",
173 f"{spec_name} is required for apps (use [] if empty)",
174 )
175 else:
176 self.warning(
177 "missing-spec",
178 f"$.{spec_name}",
179 f"{spec_name} is recommended for applets and required "
180 "for workflow compatibility",
181 )
182 continue
183 if not isinstance(value, list):
184 self.error(
185 "spec-type",
186 f"$.{spec_name}",
187 f"{spec_name} must be a JSON array",
188 )
189 continue
190 self.validate_parameter_list(spec_name, value)
192 def validate_parameter_list(
193 self,
194 spec_name: str,
195 parameters: list[Any],
196 ) -> None:
197 seen: set[str] = set()
198 is_output = spec_name == "outputSpec"
200 for index, parameter in enumerate(parameters):
201 path = f"$.{spec_name}[{index}]"
202 if not isinstance(parameter, dict):
203 self.error(
204 "parameter-type",
205 path,
206 "parameter descriptor must be a JSON object",
207 )
208 continue
210 name = parameter.get("name")
211 if not isinstance(name, str) or not NAME_RE.fullmatch(name):
212 self.error(
213 "parameter-name",
214 f"{path}.name",
215 "parameter name must match ^[A-Za-z_][A-Za-z0-9_]*$",
216 )
217 elif name in seen:
218 self.error(
219 "duplicate-parameter",
220 f"{path}.name",
221 f"duplicate parameter name {name!r}",
222 )
223 else:
224 seen.add(name)
226 class_name = parameter.get("class")
227 if not isinstance(class_name, str) or class_name not in CLASSES:
228 self.error(
229 "parameter-class",
230 f"{path}.class",
231 f"class must be one of: {', '.join(sorted(CLASSES))}",
232 )
234 if "optional" in parameter and not isinstance(parameter["optional"], bool):
235 self.error(
236 "optional-type",
237 f"{path}.optional",
238 "optional must be a boolean",
239 )
241 if is_output:
242 for unsupported in ("default", "suggestions", "choices"):
243 if unsupported in parameter:
244 self.error(
245 "output-only-field",
246 f"{path}.{unsupported}",
247 f"{unsupported} is not supported in outputSpec",
248 )
250 def validate_run_spec(self) -> None:
251 run_spec = self.manifest.get("runSpec")
252 if not isinstance(run_spec, dict):
253 self.error(
254 "missing-runspec",
255 "$.runSpec",
256 "runSpec is required and must be a JSON object",
257 )
258 return
260 has_file = isinstance(run_spec.get("file"), str) and bool(run_spec["file"])
261 has_code = isinstance(run_spec.get("code"), str) and bool(run_spec["code"])
262 if not has_file and not has_code:
263 self.error(
264 "missing-entry-source",
265 "$.runSpec",
266 "runSpec requires a non-empty file or code field",
267 )
268 elif has_file and has_code:
269 self.warning(
270 "multiple-entry-sources",
271 "$.runSpec",
272 "both file and code are supplied; keep one authoritative entry source",
273 )
275 interpreter = run_spec.get("interpreter")
276 if not isinstance(interpreter, str) or interpreter not in INTERPRETERS:
277 self.error(
278 "interpreter",
279 "$.runSpec.interpreter",
280 f"interpreter must be one of: {', '.join(sorted(INTERPRETERS))}",
281 )
283 distribution = run_spec.get("distribution")
284 if distribution != "Ubuntu":
285 self.error(
286 "distribution",
287 "$.runSpec.distribution",
288 'distribution must be "Ubuntu"',
289 )
291 release = run_spec.get("release")
292 if not isinstance(release, str) or release not in SUPPORTED_RELEASES:
293 self.error(
294 "release",
295 "$.runSpec.release",
296 "release must be 20.04 or 24.04",
297 )
298 elif release == "20.04":
299 self.warning(
300 "legacy-release",
301 "$.runSpec.release",
302 "Ubuntu 20.04 remains supported but 24.04 is preferred for new apps",
303 )
305 if run_spec.get("version") != "0":
306 self.error(
307 "aee-version",
308 "$.runSpec.version",
309 'current supported AEE version is the string "0"',
310 )
312 if "systemRequirements" in run_spec:
313 self.warning(
314 "deprecated-system-requirements",
315 "$.runSpec.systemRequirements",
316 "runSpec.systemRequirements is deprecated in dxapp.json; "
317 "use regionalOptions.<region>.systemRequirements",
318 )
320 restartable = run_spec.get("restartableEntryPoints")
321 if restartable is not None and (
322 not isinstance(restartable, str) or restartable not in {"master", "all"}
323 ):
324 self.error(
325 "restartable-entry-points",
326 "$.runSpec.restartableEntryPoints",
327 'restartableEntryPoints must be "master" or "all"',
328 )
329 if restartable == "all":
330 self.warning(
331 "restart-idempotency",
332 "$.runSpec.restartableEntryPoints",
333 "all entry points must be idempotent before enabling all",
334 )
336 self.validate_exec_depends(run_spec.get("execDepends"))
337 self.validate_execution_policy(run_spec.get("executionPolicy"))
338 self.validate_timeout_policy(run_spec.get("timeoutPolicy"))
340 def validate_exec_depends(self, dependencies: Any) -> None:
341 if dependencies is None:
342 return
343 if not isinstance(dependencies, list):
344 self.error(
345 "exec-depends-type",
346 "$.runSpec.execDepends",
347 "execDepends must be a JSON array",
348 )
349 return
351 for index, dependency in enumerate(dependencies):
352 path = f"$.runSpec.execDepends[{index}]"
353 if not isinstance(dependency, dict):
354 self.error(
355 "dependency-type",
356 path,
357 "dependency must be a JSON object",
358 )
359 continue
360 if not isinstance(dependency.get("name"), str):
361 self.error(
362 "dependency-name",
363 f"{path}.name",
364 "dependency name is required",
365 )
366 if "version" not in dependency and "tag" not in dependency:
367 self.warning(
368 "floating-dependency",
369 path,
370 "runtime dependency is not version/tag pinned; prefer a "
371 "pinned asset or bundled dependency for production",
372 )
374 def validate_execution_policy(self, policy: Any) -> None:
375 if policy is None:
376 return
377 if not isinstance(policy, dict):
378 self.error(
379 "execution-policy-type",
380 "$.runSpec.executionPolicy",
381 "executionPolicy must be a JSON object",
382 )
383 return
385 max_restarts = policy.get("maxRestarts")
386 if max_restarts is not None and (
387 not isinstance(max_restarts, int)
388 or isinstance(max_restarts, bool)
389 or not 0 <= max_restarts < 10
390 ):
391 self.error(
392 "max-restarts",
393 "$.runSpec.executionPolicy.maxRestarts",
394 "maxRestarts must be a non-negative integer below 10",
395 )
397 restart_on = policy.get("restartOn")
398 if restart_on is None:
399 return
400 if not isinstance(restart_on, dict):
401 self.error(
402 "restart-on-type",
403 "$.runSpec.executionPolicy.restartOn",
404 "restartOn must be a JSON object",
405 )
406 return
408 for reason, count in restart_on.items():
409 path = f"$.runSpec.executionPolicy.restartOn.{reason}"
410 if reason not in RESTARTABLE_FAILURES:
411 self.warning(
412 "unknown-restart-reason",
413 path,
414 "failure reason is not in the current documented restartable set",
415 )
416 if not isinstance(count, int) or isinstance(count, bool) or count < 0:
417 self.error(
418 "restart-count",
419 path,
420 "restart count must be a non-negative integer",
421 )
423 def validate_timeout_policy(self, policy: Any) -> None:
424 if policy is None:
425 return
426 if not isinstance(policy, dict):
427 self.error(
428 "timeout-policy-type",
429 "$.runSpec.timeoutPolicy",
430 "timeoutPolicy must be a JSON object",
431 )
432 return
434 for entry_point, duration in policy.items():
435 path = f"$.runSpec.timeoutPolicy.{entry_point}"
436 if not isinstance(duration, dict) or not duration:
437 self.error(
438 "timeout-duration",
439 path,
440 "timeout must be a non-empty JSON object",
441 )
442 continue
443 for unit, value in duration.items():
444 if unit not in {"days", "hours", "minutes"}:
445 self.error(
446 "timeout-unit",
447 f"{path}.{unit}",
448 "timeout units are days, hours, or minutes",
449 )
450 if (
451 not isinstance(value, (int, float))
452 or isinstance(value, bool)
453 or value < 0
454 ):
455 self.error(
456 "timeout-value",
457 f"{path}.{unit}",
458 "timeout value must be a non-negative number",
459 )
461 def validate_regional_options(self) -> None:
462 regional = self.manifest.get("regionalOptions")
463 if regional is None:
464 return
465 if not isinstance(regional, dict) or not regional:
466 self.error(
467 "regional-options-type",
468 "$.regionalOptions",
469 "regionalOptions must be a non-empty JSON object",
470 )
471 return
473 requirements_regions: set[str] = set()
474 for region, options in regional.items():
475 path = f"$.regionalOptions.{region}"
476 if not isinstance(region, str) or ":" not in region:
477 self.warning(
478 "region-name",
479 path,
480 "region should use a provider-qualified identifier such "
481 "as aws:us-east-1",
482 )
483 if not isinstance(options, dict):
484 self.error(
485 "region-options-type",
486 path,
487 "region options must be a JSON object",
488 )
489 continue
490 requirements = options.get("systemRequirements")
491 if requirements is not None:
492 requirements_regions.add(region)
493 run_spec = self.manifest.get("runSpec")
494 if isinstance(run_spec, dict) and "systemRequirements" in run_spec:
495 self.error(
496 "conflicting-system-requirements", path,
497 "regional systemRequirements cannot also be set in runSpec",
498 )
499 self.validate_system_requirements(
500 requirements,
501 f"{path}.systemRequirements",
502 )
504 if requirements_regions and len(requirements_regions) != len(regional):
505 missing = sorted(set(regional) - requirements_regions)
506 self.error(
507 "inconsistent-regional-requirements",
508 "$.regionalOptions",
509 "systemRequirements is present for only some regions; missing "
510 + ", ".join(missing),
511 )
513 def validate_system_requirements(
514 self,
515 requirements: Any,
516 path: str,
517 ) -> None:
518 if not isinstance(requirements, dict) or not requirements:
519 self.error(
520 "system-requirements-type",
521 path,
522 "systemRequirements must be a non-empty JSON object",
523 )
524 return
526 for entry_point, request in requirements.items():
527 request_path = f"{path}.{entry_point}"
528 if not isinstance(request, dict):
529 self.error(
530 "resource-request-type",
531 request_path,
532 "entry-point resource request must be a JSON object",
533 )
534 continue
535 selectors = [
536 key
537 for key in (
538 "instanceType",
539 "instanceTypeSelector",
540 "clusterSpec",
541 )
542 if key in request
543 ]
544 if "instanceTypeSelector" in selectors and len(selectors) > 1:
545 self.error(
546 "resource-selector-conflict",
547 request_path,
548 "instanceTypeSelector cannot be combined with instanceType "
549 "or clusterSpec; a fixed instanceType may accompany clusterSpec",
550 )
552 selector = request.get("instanceTypeSelector")
553 if selector is not None:
554 self.validate_instance_selector(
555 selector,
556 f"{request_path}.instanceTypeSelector",
557 )
559 def validate_instance_selector(self, selector: Any, path: str) -> None:
560 if not isinstance(selector, dict):
561 self.error(
562 "instance-selector-type",
563 path,
564 "instanceTypeSelector must be a JSON object",
565 )
566 return
567 allowed = selector.get("allowedInstanceTypes")
568 if (
569 not isinstance(allowed, list)
570 or not allowed
571 or not all(isinstance(item, str) and item for item in allowed)
572 ):
573 self.error(
574 "allowed-instance-types",
575 f"{path}.allowedInstanceTypes",
576 "allowedInstanceTypes must be a non-empty array of strings",
577 )
578 return
579 if len(set(allowed)) != len(allowed):
580 self.warning(
581 "duplicate-instance-type",
582 f"{path}.allowedInstanceTypes",
583 "duplicate instance types do not provide an additional fallback",
584 )
586 def validate_access(self) -> None:
587 access = self.manifest.get("access")
588 if access is None:
589 return
590 if not isinstance(access, dict):
591 self.error(
592 "access-type",
593 "$.access",
594 "access must be a JSON object",
595 )
596 return
598 network = access.get("network")
599 if network is not None:
600 if not isinstance(network, list) or not all(
601 isinstance(host, str) for host in network
602 ):
603 self.error(
604 "network-type",
605 "$.access.network",
606 "network must be an array of host strings",
607 )
608 elif "*" in network:
609 self.warning(
610 "broad-network",
611 "$.access.network",
612 "unrestricted network access needs explicit justification",
613 )
615 for field in ("project", "allProjects"):
616 value = access.get(field)
617 if value is not None and (
618 not isinstance(value, str) or value not in ACCESS_LEVELS
619 ):
620 self.error(
621 "access-level",
622 f"$.access.{field}",
623 f"{field} must be one of: " + ", ".join(sorted(ACCESS_LEVELS)),
624 )
626 if access.get("project") == "ADMINISTER":
627 self.warning(
628 "admin-project-access",
629 "$.access.project",
630 "ADMINISTER access is rarely needed by an analysis app",
631 )
632 if isinstance(access.get("allProjects"), str) and access["allProjects"] in ACCESS_LEVELS:
633 self.warning(
634 "all-projects-access",
635 "$.access.allProjects",
636 "cross-project access needs explicit justification",
637 )
638 if access.get("developer") is True:
639 self.warning(
640 "developer-access",
641 "$.access.developer",
642 "developer access allows advanced app operations",
643 )
645 def validate_https_app(self) -> None:
646 https_app = self.manifest.get("httpsApp")
647 if https_app is None:
648 return
649 if not isinstance(https_app, dict):
650 self.error(
651 "https-app-type",
652 "$.httpsApp",
653 "httpsApp must be a JSON object",
654 )
655 return
656 ports = https_app.get("ports")
657 if (
658 not isinstance(ports, list)
659 or not ports
660 or not all(isinstance(port, int) and port in {443, 8080, 8081} for port in ports)
661 ):
662 self.error(
663 "https-ports",
664 "$.httpsApp.ports",
665 "HTTPS app ports must be a non-empty subset of 443, 8080, and 8081",
666 )
668 def scan_for_embedded_secrets(
669 self,
670 value: Any,
671 path: str = "$",
672 ) -> None:
673 if isinstance(value, dict):
674 for key, child in value.items():
675 child_path = f"{path}.{key}"
676 if SECRET_KEY_RE.search(str(key)) and self.looks_embedded(child):
677 self.warning(
678 "embedded-secret",
679 child_path,
680 "possible credential value is embedded in dxapp.json; "
681 "use an explicit protected input or secret mechanism",
682 )
683 self.scan_for_embedded_secrets(child, child_path)
684 elif isinstance(value, list):
685 for index, child in enumerate(value):
686 self.scan_for_embedded_secrets(child, f"{path}[{index}]")
688 @staticmethod
689 def looks_embedded(value: Any) -> bool:
690 if value is None or isinstance(value, bool):
691 return False
692 if isinstance(value, str):
693 return value.strip().lower() not in PLACEHOLDER_VALUES
694 return isinstance(value, (int, float, dict, list))
697def build_parser() -> argparse.ArgumentParser:
698 parser = argparse.ArgumentParser(
699 description=__doc__,
700 formatter_class=argparse.ArgumentDefaultsHelpFormatter,
701 )
702 parser.add_argument("manifest", type=Path, help="path to dxapp.json")
703 parser.add_argument(
704 "--kind",
705 choices=("auto", "app", "applet"),
706 default="auto",
707 help="validation profile; auto treats manifests with version as apps",
708 )
709 parser.add_argument(
710 "--strict",
711 action="store_true",
712 help="exit non-zero for warnings as well as errors",
713 )
714 parser.add_argument(
715 "--json",
716 action="store_true",
717 help="emit machine-readable JSON",
718 )
719 return parser
722def load_manifest(path: Path) -> Any:
723 with path.open("r", encoding="utf-8") as handle:
724 return json.load(handle)
727def summarize(issues: Iterable[Issue]) -> dict[str, int]:
728 counts = {"error": 0, "warning": 0}
729 for issue in issues:
730 counts[issue.severity] += 1
731 return counts
734def print_text(
735 path: Path,
736 kind: str,
737 issues: list[Issue],
738) -> None:
739 counts = summarize(issues)
740 print(f"Manifest: {path}")
741 print(f"Kind: {kind}")
742 print(f"Errors: {counts['error']}")
743 print(f"Warnings: {counts['warning']}")
744 for issue in issues:
745 print(f"{issue.severity.upper()}: {issue.path}: [{issue.code}] {issue.message}")
748def main(argv: Optional[list[str]] = None) -> int:
749 parser = build_parser()
750 args = parser.parse_args(argv)
751 path = args.manifest.expanduser()
753 try:
754 manifest = load_manifest(path)
755 except (OSError, UnicodeError, json.JSONDecodeError) as error:
756 if args.json:
757 print(
758 json.dumps(
759 {
760 "manifest": str(path),
761 "kind": "unknown",
762 "valid": False,
763 "errors": 1,
764 "warnings": 0,
765 "issues": [
766 {
767 "severity": "error",
768 "code": "parse",
769 "path": "$",
770 "message": str(error),
771 }
772 ],
773 },
774 indent=2,
775 )
776 )
777 else:
778 print(f"validate_dxapp: {error}", file=sys.stderr)
779 return 2
781 validator = Validator(manifest, args.kind)
782 kind, issues = validator.validate()
783 counts = summarize(issues)
784 valid = counts["error"] == 0 and (not args.strict or counts["warning"] == 0)
786 if args.json:
787 print(
788 json.dumps(
789 {
790 "manifest": str(path),
791 "kind": kind,
792 "valid": valid,
793 "errors": counts["error"],
794 "warnings": counts["warning"],
795 "issues": [asdict(issue) for issue in issues],
796 },
797 indent=2,
798 sort_keys=True,
799 )
800 )
801 else:
802 print_text(path, kind, issues)
804 return 0 if valid else 1
807if __name__ == "__main__":
808 raise SystemExit(main())