#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ JCL Doctor V2 Professional standalone JCL linter and analyzer for MVS / z/OS style JCL. The tool reads a JCL member exported as text, normalizes continuation lines, extracts JOB, EXEC, DD, PROC, INCLUDE, control flow, datasets and inline control cards, then produces a risk score, categorized findings and optional HTML, JSON and CSV reports. No external dependency is required. """ from __future__ import annotations import argparse import csv import hashlib import html import json import re import sys from dataclasses import asdict, dataclass, field from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Tuple VERSION = "2.0.0" SEVERITY_ORDER = {"INFO": 1, "WARNING": 2, "ERROR": 3, "CRITICAL": 4} SEVERITY_POINTS = {"INFO": 1, "WARNING": 5, "ERROR": 14, "CRITICAL": 28} VALID_PROFILES = {"production", "batch", "security", "performance", "training", "strict"} SENSITIVE_PATTERNS = [ "PROD", "PAYROLL", "SALARY", "HR", "RACF", "SEC", "MASTER", "CARD", "ACCOUNT", "CUSTOMER", "CLIENT", "PII", "CONF", "SECRET", "PAY", "FINANCE", "LEDGER", ] SORT_PROGRAMS = {"SORT", "ICEMAN", "SYNCSORT", "DFSORT"} UTILITY_PROGRAMS = {"IDCAMS", "IEBGENER", "IEBCOPY", "IEFBR14", "IEHLIST", "IEHMOVE", "IKJEFT01", "IRXJCL"} COMPILER_PROGRAMS = {"IGYCRCTL", "ASMA90", "IEWL", "HEWL", "IEWBLINK", "CBL", "COBOL"} RULE_CATALOG: Dict[str, Dict[str, str]] = { "JOB_NO_CLASS": {"category": "JES", "default_severity": "WARNING", "title": "JOB missing CLASS"}, "JOB_NO_MSGCLASS": {"category": "JES", "default_severity": "WARNING", "title": "JOB missing MSGCLASS"}, "JOB_NO_NOTIFY": {"category": "Exploitability", "default_severity": "INFO", "title": "JOB missing NOTIFY"}, "JOB_NO_MSGLEVEL": {"category": "Exploitability", "default_severity": "INFO", "title": "JOB missing MSGLEVEL"}, "NO_RESTART": {"category": "Restart", "default_severity": "INFO", "title": "No restart strategy detected"}, "NO_EXEC_STEPS": {"category": "Syntax", "default_severity": "CRITICAL", "title": "No EXEC steps found"}, "DD_WITHOUT_STEP": {"category": "Syntax", "default_severity": "ERROR", "title": "DD found before EXEC"}, "EXEC_NO_TARGET": {"category": "Syntax", "default_severity": "ERROR", "title": "EXEC without program or procedure"}, "UNKNOWN_STATEMENT": {"category": "Syntax", "default_severity": "WARNING", "title": "Unknown JCL statement"}, "DD_NO_DSN_OR_TARGET": {"category": "DD", "default_severity": "ERROR", "title": "DD has no DSN/SYSOUT/DUMMY/inline data"}, "DD_NO_DISP": {"category": "DD", "default_severity": "WARNING", "title": "DD missing DISP"}, "NEW_WITHOUT_SPACE_OR_SMS": {"category": "Allocation", "default_severity": "ERROR", "title": "NEW dataset without SPACE or SMS allocation"}, "SPACE_TOO_SMALL": {"category": "Allocation", "default_severity": "WARNING", "title": "Very small SPACE allocation"}, "DISP_DELETE_RISK": {"category": "Dataset", "default_severity": "CRITICAL", "title": "Risky DELETE disposition"}, "OLD_DELETE_PROD_RISK": {"category": "Dataset", "default_severity": "CRITICAL", "title": "DISP=(OLD,DELETE) risk"}, "SENSITIVE_DSN": {"category": "Security", "default_severity": "WARNING", "title": "Sensitive-looking dataset referenced"}, "GDG_RELATIVE": {"category": "Dataset", "default_severity": "INFO", "title": "GDG relative generation detected"}, "GDG_NEW_WITHOUT_CATLG": {"category": "Dataset", "default_severity": "ERROR", "title": "New GDG generation without CATLG"}, "TEMP_DATASET_RISK": {"category": "Dataset", "default_severity": "INFO", "title": "Temporary dataset detected"}, "SORT_NO_SORTWK": {"category": "Performance", "default_severity": "WARNING", "title": "SORT step without SORTWK DD"}, "SORT_NO_SYSIN": {"category": "Syntax", "default_severity": "ERROR", "title": "SORT step without SYSIN"}, "SORT_NO_SPACE_ON_OUTPUT": {"category": "Performance", "default_severity": "WARNING", "title": "SORT output without explicit SPACE"}, "TIME_NOLIMIT": {"category": "Runtime", "default_severity": "WARNING", "title": "TIME=NOLIMIT detected"}, "NO_REGION": {"category": "Runtime", "default_severity": "INFO", "title": "No REGION specified"}, "REGION_ZERO": {"category": "Runtime", "default_severity": "WARNING", "title": "REGION=0M detected"}, "JOBLIB_USED": {"category": "Runtime", "default_severity": "INFO", "title": "JOBLIB used"}, "STEPLIB_USED": {"category": "Runtime", "default_severity": "INFO", "title": "STEPLIB used"}, "PROC_WITHOUT_JCLLIB": {"category": "Procedure", "default_severity": "WARNING", "title": "Procedure call without JCLLIB"}, "PROC_CALL_DETECTED": {"category": "Procedure", "default_severity": "INFO", "title": "Procedure call detected"}, "INCLUDE_USED": {"category": "Procedure", "default_severity": "INFO", "title": "INCLUDE member detected"}, "MIXED_COND_IF": {"category": "Control flow", "default_severity": "WARNING", "title": "COND and IF/THEN both used"}, "COND_USED": {"category": "Control flow", "default_severity": "INFO", "title": "Legacy COND control flow used"}, "IF_WITHOUT_ENDIF": {"category": "Control flow", "default_severity": "ERROR", "title": "IF without matching ENDIF"}, "ENDIF_WITHOUT_IF": {"category": "Control flow", "default_severity": "ERROR", "title": "ENDIF without matching IF"}, "LOW_COMMENT_DENSITY": {"category": "Maintainability", "default_severity": "INFO", "title": "Low comment density"}, "STEP_NO_DD": {"category": "Exploitability", "default_severity": "WARNING", "title": "Step has no DD statements"}, "IDCAMS_DELETE_PROD": {"category": "Dataset", "default_severity": "CRITICAL", "title": "IDCAMS DELETE against sensitive dataset"}, "IDCAMS_SET_MAXCC": {"category": "Control flow", "default_severity": "WARNING", "title": "IDCAMS SET MAXCC detected"}, "OUTPUT_SENSITIVE_DSN": {"category": "Security", "default_severity": "WARNING", "title": "Sensitive output dataset"}, "APF_LIBRARY_SUSPECT": {"category": "Security", "default_severity": "WARNING", "title": "Sensitive load library reference"}, "CUSTOM_BANNED_PATTERN": {"category": "Policy", "default_severity": "ERROR", "title": "Custom banned pattern matched"}, } SAMPLE_JCL_PAYROLL = r"""//PAYROLL1 JOB (ACCT),'PAYROLL NIGHTLY',CLASS=A,MSGCLASS=X,NOTIFY=&SYSUID //* PAYROLL NIGHTLY BATCH - INTENTIONAL TRAINING SAMPLE //JCLLIB ORDER=(PAYROLL.PROD.PROCLIB) //JOBLIB DD DSN=PAYROLL.PROD.LOADLIB,DISP=SHR //STEP010 EXEC PGM=IDCAMS,REGION=0M //SYSPRINT DD SYSOUT=* //SYSIN DD * DELETE PAYROLL.PROD.TEMP.KSDS CLUSTER SET MAXCC = 0 /* //STEP020 EXEC PGM=SORT //SORTIN DD DSN=PAYROLL.PROD.INPUT.G0007V00,DISP=OLD //SORTOUT DD DSN=PAYROLL.PROD.SORTED.G0007V00,DISP=(NEW,CATLG,DELETE), // SPACE=(CYL,(1,1)),UNIT=SYSDA //SYSOUT DD SYSOUT=* //SYSIN DD * SORT FIELDS=(1,10,CH,A) /* //STEP030 EXEC PGM=PAYCALC,TIME=NOLIMIT //STEPLIB DD DSN=PAYROLL.TEST.LOADLIB,DISP=SHR //PAYIN DD DSN=PAYROLL.PROD.SORTED.G0007V00,DISP=OLD //PAYOUT DD DSN=PAYROLL.PROD.RESULTS,DISP=(OLD,DELETE) //BADDD DD DISP=SHR //REPORT DD SYSOUT=* //SYSUDUMP DD SYSOUT=* """ SAMPLE_JCL_CLEAN = r"""//INVOICE1 JOB (ACCT),'INVOICE DAILY',CLASS=B,MSGCLASS=X,MSGLEVEL=(1,1),NOTIFY=&SYSUID,RESTART=STEP020 //* INVOICE DAILY BATCH //* RESTART POINTS: STEP010, STEP020, STEP030 //JCLLIB ORDER=(CORP.PROD.PROCLIB) //STEP010 EXEC PGM=IEFBR14,REGION=0M //ALLOC1 DD DSN=CORP.PROD.INVOICE.WORK(+1),DISP=(NEW,CATLG,DELETE), // SPACE=(CYL,(50,25)),UNIT=SYSDA //STEP020 EXEC PGM=SORT,REGION=64M //SORTIN DD DSN=CORP.PROD.INVOICE.INPUT(0),DISP=SHR //SORTOUT DD DSN=CORP.PROD.INVOICE.SORTED(+1),DISP=(NEW,CATLG,DELETE), // SPACE=(CYL,(100,50)),UNIT=SYSDA //SORTWK01 DD UNIT=SYSDA,SPACE=(CYL,(200,100)) //SORTWK02 DD UNIT=SYSDA,SPACE=(CYL,(200,100)) //SYSOUT DD SYSOUT=* //SYSIN DD * SORT FIELDS=(1,20,CH,A) /* //STEP030 EXEC PGM=INVPOST,REGION=128M //STEPLIB DD DSN=CORP.PROD.LOADLIB,DISP=SHR //INPUT DD DSN=CORP.PROD.INVOICE.SORTED(+1),DISP=SHR //REPORT DD SYSOUT=* """ SAMPLE_JCL_PROC = r"""//PROCJOB1 JOB (ACCT),'PROC TEST',MSGCLASS=X //* SAMPLE WITH PROCEDURE CALLS AND INCLUDE //STEP001 EXEC PAYPROC,REGION=32M //INFILE DD DSN=FINANCE.PROD.INPUT(0),DISP=SHR //OUTFILE DD DSN=FINANCE.PROD.OUTPUT(+1),DISP=(NEW,KEEP,DELETE) //INCLUDE MEMBER=SITEPARM // IF (STEP001.RC > 4) THEN //STEP002 EXEC PGM=IEFBR14 //ENDIF """ SAMPLE_JCL_SECURITY = r"""//SECCHK1 JOB (SEC),'SECURITY AUDIT',CLASS=A,MSGCLASS=X //* INTENTIONAL SECURITY POLICY SAMPLE //STEP010 EXEC PGM=IDCAMS,TIME=NOLIMIT //SYSPRINT DD SYSOUT=* //SYSIN DD * DELETE SYS1.RACF.BACKUP CLUSTER SET MAXCC=0 /* //STEP020 EXEC PGM=IEBGENER //SYSUT1 DD DSN=HR.PROD.SALARY.MASTER,DISP=OLD //SYSUT2 DD DSN=HR.PROD.SALARY.COPY,DISP=(OLD,DELETE) //SYSPRINT DD SYSOUT=* //SYSIN DD DUMMY """ SAMPLES = { "payroll": SAMPLE_JCL_PAYROLL, "clean": SAMPLE_JCL_CLEAN, "proc": SAMPLE_JCL_PROC, "security": SAMPLE_JCL_SECURITY, } @dataclass class JCLStatement: line_no: int raw: str name: str = "" operation: str = "" parameters: str = "" continued: bool = False @dataclass class JCLJob: name: str = "UNKNOWN" accounting: str = "" programmer: str = "" parameters: Dict[str, str] = field(default_factory=dict) line_no: int = 0 @dataclass class JCLStep: name: str program: str = "" procedure: str = "" parameters: Dict[str, str] = field(default_factory=dict) line_no: int = 0 dd_count: int = 0 is_sort: bool = False is_utility: bool = False is_compiler: bool = False uses_proc: bool = False @dataclass class JCLDD: step_name: str name: str parameters: Dict[str, str] = field(default_factory=dict) raw_parameters: str = "" line_no: int = 0 is_inline: bool = False is_sysout: bool = False is_dummy: bool = False dataset: str = "" disposition: str = "" unit: str = "" space: str = "" dcb: str = "" sms_classes: Dict[str, str] = field(default_factory=dict) @dataclass class DatasetRef: dataset: str step_name: str dd_name: str disposition: str = "" access: str = "" is_gdg: bool = False is_temp: bool = False is_sensitive: bool = False is_output: bool = False line_no: int = 0 @dataclass class InlineBlock: step_name: str dd_name: str line_no: int lines: List[str] = field(default_factory=list) @dataclass class Finding: severity: str rule_id: str category: str title: str message: str recommendation: str confidence: str = "MEDIUM" line_no: int = 0 object_type: str = "" object_name: str = "" evidence: str = "" @dataclass class CategoryScore: category: str finding_count: int risk_points: int highest_severity: str @dataclass class JCLComparison: baseline_source: str = "" baseline_risk_score: int = 0 current_risk_score: int = 0 delta_risk_score: int = 0 baseline_finding_count: int = 0 current_finding_count: int = 0 delta_finding_count: int = 0 new_rule_ids: List[str] = field(default_factory=list) resolved_rule_ids: List[str] = field(default_factory=list) @dataclass class JCLAnalysis: source_name: str profile: str status: str highest_severity: str risk_score: int job: JCLJob steps: List[JCLStep] dds: List[JCLDD] datasets: List[DatasetRef] inline_blocks: List[InlineBlock] findings: List[Finding] category_scores: List[CategoryScore] comparison: Optional[JCLComparison] metrics: Dict[str, Any] def normalize_spaces(value: str) -> str: return re.sub(r"\s+", " ", value.strip()) def strip_outer_parentheses(value: str) -> str: value = value.strip() if value.startswith("(") and value.endswith(")"): depth = 0 for index, char in enumerate(value): if char == "(": depth += 1 elif char == ")": depth -= 1 if depth == 0 and index != len(value) - 1: return value return value[1:-1] return value def split_top_level_commas(value: str) -> List[str]: parts: List[str] = [] current: List[str] = [] depth = 0 quote: Optional[str] = None for char in value: if quote: current.append(char) if char == quote: quote = None continue if char in ("'", '"'): quote = char current.append(char) continue if char == "(": depth += 1 current.append(char) continue if char == ")": depth = max(depth - 1, 0) current.append(char) continue if char == "," and depth == 0: part = "".join(current).strip() if part: parts.append(part) current = [] continue current.append(char) tail = "".join(current).strip() if tail: parts.append(tail) return parts def parse_keyword_parameters(raw: str) -> Dict[str, str]: params: Dict[str, str] = {} positional: List[str] = [] for part in split_top_level_commas(raw): if "=" in part: key, value = part.split("=", 1) params[key.strip().upper()] = value.strip() else: positional.append(part.strip()) if positional: params["_POSITIONAL"] = ",".join(positional) return params def is_jcl_continuation(line: str, current: str) -> bool: if not line.startswith("//") or line.startswith("//*"): return False body = line[2:] if not body: return False # Common continuation: name and operation fields are blank after //. return bool(current and re.match(r"^\s{2,}\S", body)) def unfold_jcl_lines(text: str) -> List[Tuple[int, str]]: unfolded: List[Tuple[int, str]] = [] current_line_no = 0 current = "" for idx, raw in enumerate(text.splitlines(), start=1): line = raw.rstrip("\n") if not line.strip(): continue if line.startswith("//") and not line.startswith("//*"): if is_jcl_continuation(line, current): continuation = line[2:].strip() if current: current = current.rstrip().rstrip(",") + "," + continuation else: current_line_no = idx current = line continue if current: unfolded.append((current_line_no, current)) current_line_no = idx current = line else: if current: unfolded.append((current_line_no, current)) current = "" unfolded.append((idx, line)) if current: unfolded.append((current_line_no, current)) return unfolded def parse_jcl_statement(line_no: int, line: str) -> Optional[JCLStatement]: raw = line.rstrip() if raw.startswith("//*"): return JCLStatement(line_no=line_no, raw=raw, operation="COMMENT", parameters=raw[3:].strip()) if not raw.startswith("//"): return JCLStatement(line_no=line_no, raw=raw, operation="DATA", parameters=raw.strip()) body = raw[2:].rstrip() if not body.strip(): return None if body.lstrip().startswith("*"): return JCLStatement(line_no=line_no, raw=raw, operation="COMMENT", parameters=body.strip()[1:].strip()) match = re.match(r"^(?P[A-Za-z0-9$#@_]+)?\s*(?PJOB|EXEC|DD|PROC|PEND|INCLUDE|JCLLIB|SET|IF|THEN|ELSE|ENDIF|OUTPUT|XMIT)\b\s*(?P.*)$", body, re.IGNORECASE) if not match: # Fallback for control statements such as // IF (...) match2 = re.match(r"^(?PIF|ELSE|ENDIF)\b\s*(?P.*)$", body.strip(), re.IGNORECASE) if match2: return JCLStatement(line_no=line_no, raw=raw, operation=match2.group("op").upper(), parameters=(match2.group("params") or "").strip()) return JCLStatement(line_no=line_no, raw=raw, operation="UNKNOWN", parameters=body.strip()) return JCLStatement( line_no=line_no, raw=raw, name=(match.group("name") or "").upper(), operation=(match.group("op") or "").upper(), parameters=(match.group("params") or "").strip(), continued=raw.rstrip().endswith(","), ) def parse_statements(text: str) -> List[JCLStatement]: statements = [parse_jcl_statement(no, line) for no, line in unfold_jcl_lines(text)] return [stmt for stmt in statements if stmt is not None] def extract_inline_blocks(statements: List[JCLStatement]) -> List[InlineBlock]: blocks: List[InlineBlock] = [] current_step = "" current_block: Optional[InlineBlock] = None for stmt in statements: if stmt.operation == "EXEC": current_step = stmt.name or f"STEP{len([b for b in blocks if b.step_name == current_step]) + 1:03d}" if current_block: current_block = None continue if stmt.operation == "DD": params = parse_keyword_parameters(stmt.parameters) raw_upper = stmt.parameters.strip().upper() positional = params.get("_POSITIONAL", "").upper() inline_marker = raw_upper.startswith("*") or raw_upper.startswith("DATA") or positional in {"*", "DATA"} or positional.startswith("*,") if inline_marker: current_block = InlineBlock(step_name=current_step, dd_name=stmt.name or "", line_no=stmt.line_no) blocks.append(current_block) else: current_block = None continue if stmt.operation == "DATA": if stmt.parameters.strip() == "/*": current_block = None elif current_block: current_block.lines.append(stmt.raw) continue if stmt.operation not in {"COMMENT"}: current_block = None return blocks def parse_job_params(params: str) -> Tuple[str, str, Dict[str, str]]: pieces = split_top_level_commas(params) accounting = pieces[0] if pieces else "" programmer = "" rest: List[str] if len(pieces) > 1 and "=" not in pieces[1]: programmer = pieces[1] rest = pieces[2:] else: rest = pieces[1:] param_dict = parse_keyword_parameters(",".join(rest)) return accounting, programmer, param_dict def disp_actions(value: str) -> List[str]: clean = strip_outer_parentheses(value.upper()).replace(" ", "") if not clean: return [] return [part for part in split_top_level_commas(clean) if part] def severity_at_least(value: str, threshold: str) -> bool: return SEVERITY_ORDER.get(value.upper(), 0) >= SEVERITY_ORDER.get(threshold.upper(), 999) def adjust_severity(rule_id: str, severity: str, profile: str) -> str: profile = profile.lower() rank = SEVERITY_ORDER.get(severity, 1) if profile == "training": if rule_id in {"SENSITIVE_DSN", "LOW_COMMENT_DENSITY", "NO_REGION", "JOB_NO_NOTIFY"}: rank = max(1, rank - 1) if profile == "strict": if severity in {"INFO", "WARNING"}: rank = min(4, rank + 1) if profile == "security": if rule_id in {"SENSITIVE_DSN", "OUTPUT_SENSITIVE_DSN", "APF_LIBRARY_SUSPECT", "CUSTOM_BANNED_PATTERN"}: rank = min(4, rank + 1) if profile == "performance": if rule_id in {"SORT_NO_SORTWK", "SORT_NO_SPACE_ON_OUTPUT", "SPACE_TOO_SMALL", "REGION_ZERO"}: rank = min(4, rank + 1) if profile == "production": if rule_id in {"OLD_DELETE_PROD_RISK", "DISP_DELETE_RISK", "IDCAMS_DELETE_PROD"}: rank = 4 for sev, value in SEVERITY_ORDER.items(): if value == rank: return sev return severity def redact_dataset_name(value: str) -> str: if not value: return value return re.sub(r"\b[A-Z0-9$#@][A-Z0-9$#@.-]{2,}(?:\([+-]?\d+\))?\b", "", value.upper()) def maybe_redact(value: str, redact: bool) -> str: return redact_dataset_name(value) if redact else value def is_sensitive_dataset(dsn: str, extra_patterns: Optional[List[str]] = None) -> bool: upper = dsn.upper() patterns = list(SENSITIVE_PATTERNS) if extra_patterns: patterns.extend(extra_patterns) return any(pattern.upper() in upper for pattern in patterns) def is_gdg_reference(dsn: str) -> bool: return bool(re.search(r"\([+-]?\d+\)$", dsn.strip())) def is_temp_dataset(dsn: str) -> bool: return dsn.strip().startswith("&&") def is_output_disp(actions: List[str]) -> bool: if not actions: return False return actions[0] in {"NEW", "MOD"} def has_sms_allocation(params: Dict[str, str]) -> bool: return any(key in params for key in {"DATACLAS", "STORCLAS", "MGMTCLAS", "LIKE"}) def parse_space_primary_secondary(space: str) -> Optional[Tuple[str, int, int]]: clean = space.upper().replace(" ", "") match = re.search(r"\((TRK|CYL|BLKLGTH|\d+),\((\d+),(\d+)\)", clean) if not match: return None return match.group(1), int(match.group(2)), int(match.group(3)) def fingerprint_finding(f: Finding) -> str: base = f"{f.rule_id}|{f.object_type}|{f.object_name}|{f.line_no}|{f.title}" return hashlib.sha1(base.encode("utf-8", errors="ignore")).hexdigest()[:12] def load_custom_policy(path: Optional[str]) -> Dict[str, Any]: if not path: return {} data = json.loads(Path(path).read_text(encoding="utf-8")) if not isinstance(data, dict): raise ValueError("Custom policy must be a JSON object") return data def default_rule_metadata(rule_id: str) -> Dict[str, str]: return RULE_CATALOG.get(rule_id, {"category": "General", "default_severity": "INFO", "title": rule_id}) def compute_category_scores(findings: List[Finding]) -> List[CategoryScore]: by_category: Dict[str, List[Finding]] = {} for finding in findings: by_category.setdefault(finding.category, []).append(finding) scores: List[CategoryScore] = [] for category, items in by_category.items(): points = sum(SEVERITY_POINTS.get(item.severity, 1) for item in items) highest = max(items, key=lambda item: SEVERITY_ORDER.get(item.severity, 0)).severity if items else "INFO" scores.append(CategoryScore(category=category, finding_count=len(items), risk_points=points, highest_severity=highest)) return sorted(scores, key=lambda item: item.risk_points, reverse=True) def compare_analyses(current: "JCLAnalysis", baseline: "JCLAnalysis") -> JCLComparison: current_keys = {f.rule_id for f in current.findings} baseline_keys = {f.rule_id for f in baseline.findings} return JCLComparison( baseline_source=baseline.source_name, baseline_risk_score=baseline.risk_score, current_risk_score=current.risk_score, delta_risk_score=current.risk_score - baseline.risk_score, baseline_finding_count=len(baseline.findings), current_finding_count=len(current.findings), delta_finding_count=len(current.findings) - len(baseline.findings), new_rule_ids=sorted(current_keys - baseline_keys), resolved_rule_ids=sorted(baseline_keys - current_keys), ) def analyze_jcl( text: str, source_name: str = "stdin", profile: str = "production", redact: bool = False, custom_policy: Optional[Dict[str, Any]] = None, comparison: Optional[JCLComparison] = None, ) -> JCLAnalysis: if profile not in VALID_PROFILES: raise ValueError(f"Invalid profile: {profile}") custom_policy = custom_policy or {} extra_sensitive_patterns = custom_policy.get("sensitive_patterns", []) if isinstance(custom_policy.get("sensitive_patterns", []), list) else [] banned_patterns = custom_policy.get("banned_patterns", []) if isinstance(custom_policy.get("banned_patterns", []), list) else [] statements = parse_statements(text) inline_blocks = extract_inline_blocks(statements) job = JCLJob() steps: List[JCLStep] = [] dds: List[JCLDD] = [] datasets: List[DatasetRef] = [] findings: List[Finding] = [] current_step: Optional[JCLStep] = None comment_count = 0 has_restart = False has_jcllib = False has_joblib = False has_steplib = False has_cond = False has_if_then = False has_region_on_any_step = False include_count = 0 if_stack: List[int] = [] sort_steps: List[str] = [] sortwk_by_step: Dict[str, int] = {} sysin_by_step: Dict[str, int] = {} sortout_by_step: Dict[str, JCLDD] = {} proc_calls: List[JCLStep] = [] def add_finding( rule_id: str, message: str, recommendation: str, stmt: Optional[JCLStatement] = None, object_type: str = "", object_name: str = "", confidence: str = "MEDIUM", severity_override: Optional[str] = None, evidence: Optional[str] = None, ) -> None: meta = default_rule_metadata(rule_id) severity = severity_override or meta.get("default_severity", "INFO") severity = adjust_severity(rule_id, severity, profile) findings.append(Finding( severity=severity, rule_id=rule_id, category=meta.get("category", "General"), title=meta.get("title", rule_id), message=maybe_redact(message, redact), recommendation=recommendation, confidence=confidence, line_no=stmt.line_no if stmt else 0, object_type=object_type, object_name=object_name, evidence=maybe_redact(evidence if evidence is not None else (stmt.raw if stmt else ""), redact), )) for stmt in statements: op = stmt.operation params_upper = stmt.parameters.upper() if op == "COMMENT": comment_count += 1 continue if op == "DATA": continue for pattern in banned_patterns: if pattern and re.search(pattern, stmt.raw, re.IGNORECASE): add_finding( "CUSTOM_BANNED_PATTERN", f"Custom banned policy pattern matched: {pattern}", "Review the custom policy rule and update the JCL before production submission.", stmt, "LINE", str(stmt.line_no), confidence="HIGH", ) if op == "JOB": accounting, programmer, job_params = parse_job_params(stmt.parameters) job = JCLJob(name=stmt.name or "UNKNOWN", accounting=accounting, programmer=programmer, parameters=job_params, line_no=stmt.line_no) has_restart = "RESTART" in job_params if "NOTIFY" not in job_params: add_finding("JOB_NO_NOTIFY", "The JOB statement does not specify NOTIFY.", "Consider using NOTIFY so the submitter or operations team receives completion messages.", stmt, "JOB", job.name) if "MSGCLASS" not in job_params: add_finding("JOB_NO_MSGCLASS", "The JOB statement does not specify MSGCLASS.", "Set MSGCLASS according to the site output routing policy.", stmt, "JOB", job.name) if "CLASS" not in job_params: add_finding("JOB_NO_CLASS", "The JOB statement does not specify CLASS.", "Set CLASS explicitly to avoid scheduler or JES default surprises.", stmt, "JOB", job.name) if "MSGLEVEL" not in job_params: add_finding("JOB_NO_MSGLEVEL", "The JOB statement does not specify MSGLEVEL.", "Set MSGLEVEL to capture the required diagnostic information for production support.", stmt, "JOB", job.name) continue if op == "JCLLIB": has_jcllib = True continue if op == "INCLUDE": include_count += 1 add_finding("INCLUDE_USED", f"INCLUDE statement detected: {stmt.parameters}", "Ensure included members are version-controlled and available in the execution environment.", stmt, "INCLUDE", stmt.name or "INCLUDE") continue if op == "EXEC": params = parse_keyword_parameters(stmt.parameters) program = params.get("PGM", "").upper() positional = params.get("_POSITIONAL", "").upper() procedure = params.get("PROC", "").upper() or (positional if not program else "") step_name = stmt.name or f"STEP{len(steps) + 1:03d}" current_step = JCLStep( name=step_name, program=program, procedure=procedure, parameters=params, line_no=stmt.line_no, is_sort=program in SORT_PROGRAMS, is_utility=program in UTILITY_PROGRAMS, is_compiler=program in COMPILER_PROGRAMS, uses_proc=bool(procedure and not program), ) steps.append(current_step) if current_step.uses_proc: proc_calls.append(current_step) add_finding("PROC_CALL_DETECTED", f"Step {step_name} calls procedure {procedure}.", "Verify procedure version, override points and JCLLIB resolution.", stmt, "STEP", step_name) if "REGION" in params: has_region_on_any_step = True if params.get("REGION", "").replace(" ", "").upper() in {"0M", "0K"}: add_finding("REGION_ZERO", f"Step {step_name} uses REGION={params.get('REGION')}." , "Confirm that unrestricted or very large region is accepted by the site policy.", stmt, "STEP", step_name) if "COND" in params: has_cond = True add_finding("COND_USED", f"Step {step_name} uses COND logic.", "Document step bypass conditions and prefer clear IF/THEN when possible.", stmt, "STEP", step_name) if "TIME" in params and "NOLIMIT" in params.get("TIME", "").upper(): add_finding("TIME_NOLIMIT", f"Step {step_name} uses TIME=NOLIMIT.", "Confirm that the job is not able to loop indefinitely and that operations accepts this policy.", stmt, "STEP", step_name) if not program and not procedure: add_finding("EXEC_NO_TARGET", f"Step {step_name} does not clearly specify PGM= or PROC.", "Specify PGM=program or a valid procedure name.", stmt, "STEP", step_name) if program in SORT_PROGRAMS: sort_steps.append(step_name) continue if op == "IF": has_if_then = True if_stack.append(stmt.line_no) continue if op == "ELSE": if not if_stack: add_finding("ENDIF_WITHOUT_IF", "ELSE statement found without a currently open IF.", "Review conditional control flow and matching IF/ENDIF statements.", stmt, "CONTROL", "ELSE") continue if op == "ENDIF": if not if_stack: add_finding("ENDIF_WITHOUT_IF", "ENDIF statement found without matching IF.", "Review conditional control flow and matching IF/ENDIF statements.", stmt, "CONTROL", "ENDIF") else: if_stack.pop() continue if op == "DD": dd_name_peek = (stmt.name or "").upper() if current_step is None: if dd_name_peek in {"JOBLIB", "JOBCAT"}: step_name = "" else: add_finding("DD_WITHOUT_STEP", f"DD {stmt.name} appears before any EXEC statement.", "Move the DD under the correct EXEC step or review the JCL structure.", stmt, "DD", stmt.name) step_name = "" else: step_name = current_step.name current_step.dd_count += 1 params = parse_keyword_parameters(stmt.parameters) raw_params = stmt.parameters positional = params.get("_POSITIONAL", "") is_inline = raw_params.strip().startswith("*") or positional.upper() in {"*", "DATA"} or positional.upper().startswith("*,") is_sysout = "SYSOUT" in params is_dummy = "DUMMY" in positional.upper() or "DUMMY" in raw_params.upper() dsn = params.get("DSN", "").strip() disp = params.get("DISP", "") unit = params.get("UNIT", "") space = params.get("SPACE", "") dcb = params.get("DCB", "") sms_classes = {key: params[key] for key in ("DATACLAS", "STORCLAS", "MGMTCLAS", "LIKE") if key in params} dd_name = stmt.name or "" dd = JCLDD( step_name=step_name, name=dd_name, parameters=params, raw_parameters=raw_params, line_no=stmt.line_no, is_inline=is_inline, is_sysout=is_sysout, is_dummy=is_dummy, dataset=maybe_redact(dsn, redact), disposition=disp, unit=unit, space=space, dcb=dcb, sms_classes=sms_classes, ) dds.append(dd) if dd_name.upper() == "JOBLIB": has_joblib = True if dd_name.upper() == "STEPLIB": has_steplib = True if dd_name.upper().startswith("SORTWK"): sortwk_by_step[step_name] = sortwk_by_step.get(step_name, 0) + 1 if dd_name.upper() == "SYSIN": sysin_by_step[step_name] = sysin_by_step.get(step_name, 0) + 1 if dd_name.upper() in {"SORTOUT", "OUT", "OUTPUT", "SYSUT2"}: sortout_by_step[step_name] = dd actions = disp_actions(disp) if disp else [] if dsn: dataset_ref = DatasetRef( dataset=maybe_redact(dsn, redact), step_name=step_name, dd_name=dd_name, disposition=disp, access=actions[0] if actions else "", is_gdg=is_gdg_reference(dsn), is_temp=is_temp_dataset(dsn), is_sensitive=is_sensitive_dataset(dsn, extra_sensitive_patterns), is_output=is_output_disp(actions), line_no=stmt.line_no, ) datasets.append(dataset_ref) if dataset_ref.is_sensitive: add_finding("SENSITIVE_DSN", f"DD {dd_name} references dataset {dsn}.", "Validate that DISP and access permissions match production data handling policy.", stmt, "DD", f"{step_name}.{dd_name}", confidence="HIGH") if dataset_ref.is_output and dataset_ref.is_sensitive: add_finding("OUTPUT_SENSITIVE_DSN", f"DD {dd_name} writes or modifies sensitive-looking dataset {dsn}.", "Confirm that this output is authorized and covered by retention/security policy.", stmt, "DD", f"{step_name}.{dd_name}", confidence="HIGH") if dataset_ref.is_gdg: add_finding("GDG_RELATIVE", f"Dataset {dsn} uses a relative generation reference.", "Verify that the referenced generation exists at execution time and that scheduler dependencies are correct.", stmt, "DD", f"{step_name}.{dd_name}") if actions and actions[0] == "NEW" and "CATLG" not in actions: add_finding("GDG_NEW_WITHOUT_CATLG", f"New GDG generation {dsn} is not cataloged by DISP.", "Use DISP=(NEW,CATLG,DELETE) unless a deliberate non-cataloged generation is required.", stmt, "DD", f"{step_name}.{dd_name}") if dataset_ref.is_temp: add_finding("TEMP_DATASET_RISK", f"Temporary dataset {dsn} detected.", "Confirm temporary dataset lifecycle and step dependencies.", stmt, "DD", f"{step_name}.{dd_name}") if any(token in dsn.upper() for token in {"APF", "LINKLIB", "LNKLST", "LOADLIB", "AUTHLIB", "SYS1."}): add_finding("APF_LIBRARY_SUSPECT", f"DD {dd_name} references sensitive load/library dataset {dsn}.", "Confirm library order, APF requirements and that test libraries cannot shadow production modules.", stmt, "DD", f"{step_name}.{dd_name}") if not dsn and not is_sysout and not is_dummy and not is_inline and "UNIT" not in params: add_finding("DD_NO_DSN_OR_TARGET", f"DD {dd_name} in step {step_name} has no clear target.", "Add DSN=, SYSOUT=*, DUMMY, inline data marker, or a valid allocation statement as appropriate.", stmt, "DD", f"{step_name}.{dd_name}") if disp: if "DELETE" in actions and not any(action in {"NEW", "MOD"} for action in actions[:1]): add_finding("DISP_DELETE_RISK", f"DD {dd_name} uses DISP={disp} for dataset {dsn or ''}.", "Review whether this dataset may be deleted on normal or abnormal termination.", stmt, "DD", f"{step_name}.{dd_name}", confidence="HIGH") if actions and actions[0] == "OLD" and "DELETE" in actions: add_finding("OLD_DELETE_PROD_RISK", f"DD {dd_name} uses OLD with DELETE disposition.", "Avoid deleting existing datasets unless this is an explicitly controlled cleanup step.", stmt, "DD", f"{step_name}.{dd_name}", confidence="HIGH") if actions and actions[0] == "NEW" and "SPACE" not in params and not has_sms_allocation(params) and not is_sysout: add_finding("NEW_WITHOUT_SPACE_OR_SMS", f"DD {dd_name} creates a NEW dataset without SPACE= or SMS class/LIKE allocation.", "Specify SPACE or SMS allocation attributes such as DATACLAS/STORCLAS/MGMTCLAS/LIKE.", stmt, "DD", f"{step_name}.{dd_name}") elif dsn and not is_inline: add_finding("DD_NO_DISP", f"DD {dd_name} references a dataset without DISP=.", "Add DISP explicitly to avoid site default ambiguity.", stmt, "DD", f"{step_name}.{dd_name}") if space: parsed_space = parse_space_primary_secondary(space) if parsed_space: unit_type, primary, secondary = parsed_space if primary <= 1 and secondary <= 1: add_finding("SPACE_TOO_SMALL", f"DD {dd_name} uses SPACE={space}.", "Validate the expected output volume and adjust primary/secondary allocation.", stmt, "DD", f"{step_name}.{dd_name}") continue if op in {"PROC", "PEND", "SET", "OUTPUT", "XMIT"}: continue if op == "UNKNOWN": add_finding("UNKNOWN_STATEMENT", f"The parser could not classify line {stmt.line_no}.", "Review the syntax or add a site-specific parser rule.", stmt, "LINE", str(stmt.line_no)) inline_by_key = {(block.step_name, block.dd_name): block for block in inline_blocks} for block in inline_blocks: joined = "\n".join(block.lines).upper() evidence = "\n".join(block.lines[:5]) if "SET MAXCC" in joined: add_finding("IDCAMS_SET_MAXCC", f"Inline control cards in {block.step_name}.{block.dd_name} contain SET MAXCC.", "Confirm that non-zero utility conditions are not being hidden from production monitoring.", None, "DD", f"{block.step_name}.{block.dd_name}", evidence=evidence) if re.search(r"\bDELETE\s+[A-Z0-9$#@.]+", joined): match = re.search(r"\bDELETE\s+([A-Z0-9$#@.]+)", joined) target = match.group(1) if match else "" if is_sensitive_dataset(target, extra_sensitive_patterns): add_finding("IDCAMS_DELETE_PROD", f"Inline control cards delete sensitive-looking dataset {target}.", "Require explicit change approval and verify that this is not a production destructive action.", None, "DD", f"{block.step_name}.{block.dd_name}", confidence="HIGH", evidence=evidence) if not steps: add_finding("NO_EXEC_STEPS", "No EXEC statements were detected in the JCL.", "Verify the input file and make sure it contains a complete JCL member.", None, "JOB", job.name) for step in steps: if step.dd_count == 0: add_finding("STEP_NO_DD", f"Step {step.name} has no DD statements.", "Confirm that this is intentional and not a missing procedure expansion.", None, "STEP", step.name) for sort_step in sort_steps: if sortwk_by_step.get(sort_step, 0) == 0: add_finding("SORT_NO_SORTWK", f"SORT-like step {sort_step} has no explicit SORTWK DD.", "For large sorts, consider explicit work datasets or verify DFSORT dynamic allocation policy.", None, "STEP", sort_step) if sysin_by_step.get(sort_step, 0) == 0: add_finding("SORT_NO_SYSIN", f"SORT-like step {sort_step} has no SYSIN DD.", "Provide SORT control statements or verify that they are supplied through a procedure.", None, "STEP", sort_step) out_dd = sortout_by_step.get(sort_step) if out_dd and not out_dd.space and not out_dd.sms_classes and not out_dd.is_sysout: add_finding("SORT_NO_SPACE_ON_OUTPUT", f"SORT output DD {out_dd.name} in step {sort_step} has no explicit SPACE or SMS allocation.", "For predictable large sorts, define SPACE or SMS class/LIKE allocation.", None, "DD", f"{sort_step}.{out_dd.name}") if not has_restart: add_finding("NO_RESTART", "The JOB statement does not include RESTART and no restart metadata was detected.", "For production batch jobs, document restart points and restart granularity.", None, "JOB", job.name) if not has_region_on_any_step: add_finding("NO_REGION", "No step-level REGION parameter was detected.", "Review site defaults and specify REGION for memory-sensitive jobs.", None, "JOB", job.name) if has_joblib: add_finding("JOBLIB_USED", "JOBLIB affects load library resolution for the whole job.", "Confirm that the library order is controlled and does not shadow production modules unexpectedly.", None, "JOB", job.name) if has_steplib: add_finding("STEPLIB_USED", "STEPLIB affects load library resolution for a step.", "Confirm that the library order is controlled and matches deployment policy.", None, "JOB", job.name) if comment_count < max(2, len(steps)) and profile in {"production", "batch", "strict"}: add_finding("LOW_COMMENT_DENSITY", "The JCL contains few comments compared with the number of steps.", "Add operational comments explaining job purpose, restart points, and critical datasets.", None, "JOB", job.name) if has_cond and has_if_then: add_finding("MIXED_COND_IF", "The JCL appears to mix legacy COND logic and IF/THEN logic.", "Prefer one clear control flow style when possible and document execution paths.", None, "JOB", job.name) if if_stack: add_finding("IF_WITHOUT_ENDIF", f"{len(if_stack)} IF statement(s) were not closed by ENDIF.", "Review conditional control flow and add matching ENDIF statements.", None, "CONTROL", job.name) if not has_jcllib and proc_calls: add_finding("PROC_WITHOUT_JCLLIB", "A procedure appears to be called but no JCLLIB ORDER statement was detected.", "Ensure the procedure library is resolved by site defaults or add JCLLIB ORDER explicitly.", None, "JOB", job.name) max_sev = "INFO" for finding in findings: if SEVERITY_ORDER[finding.severity] > SEVERITY_ORDER[max_sev]: max_sev = finding.severity raw_score = sum(SEVERITY_POINTS.get(f.severity, 1) for f in findings) category_scores = compute_category_scores(findings) # High-risk diversity should matter, but keep score bounded. diversity_bonus = min(12, max(0, len([c for c in category_scores if c.finding_count > 0]) - 2) * 2) risk_score = min(100, raw_score + diversity_bonus) status = "BLOCKED" if max_sev == "CRITICAL" else "NEEDS_REVIEW" if max_sev in {"ERROR", "WARNING"} else "OK" metrics = { "version": VERSION, "statement_count": len(statements), "comment_count": comment_count, "step_count": len(steps), "dd_count": len(dds), "dataset_count": len(datasets), "inline_block_count": len(inline_blocks), "include_count": include_count, "proc_call_count": len(proc_calls), "sensitive_dataset_count": sum(1 for item in datasets if item.is_sensitive), "gdg_count": sum(1 for item in datasets if item.is_gdg), "temp_dataset_count": sum(1 for item in datasets if item.is_temp), "output_dataset_count": sum(1 for item in datasets if item.is_output), "finding_count": len(findings), "critical_count": sum(1 for f in findings if f.severity == "CRITICAL"), "error_count": sum(1 for f in findings if f.severity == "ERROR"), "warning_count": sum(1 for f in findings if f.severity == "WARNING"), "info_count": sum(1 for f in findings if f.severity == "INFO"), "has_restart": has_restart, "has_jcllib": has_jcllib, "has_joblib": has_joblib, "has_steplib": has_steplib, "has_cond": has_cond, "has_if_then": has_if_then, } return JCLAnalysis( source_name=source_name, profile=profile, status=status, highest_severity=max_sev, risk_score=risk_score, job=job, steps=steps, dds=dds, datasets=datasets, inline_blocks=inline_blocks, findings=findings, category_scores=category_scores, comparison=comparison, metrics=metrics, ) def analysis_to_dict(analysis: JCLAnalysis) -> Dict[str, Any]: return { "source_name": analysis.source_name, "profile": analysis.profile, "status": analysis.status, "highest_severity": analysis.highest_severity, "risk_score": analysis.risk_score, "job": asdict(analysis.job), "steps": [asdict(step) for step in analysis.steps], "dds": [asdict(dd) for dd in analysis.dds], "datasets": [asdict(item) for item in analysis.datasets], "inline_blocks": [asdict(item) for item in analysis.inline_blocks], "findings": [asdict(finding) for finding in analysis.findings], "category_scores": [asdict(score) for score in analysis.category_scores], "comparison": asdict(analysis.comparison) if analysis.comparison else None, "metrics": analysis.metrics, } def write_json(analysis: JCLAnalysis, path: str) -> None: Path(path).write_text(json.dumps(analysis_to_dict(analysis), indent=2), encoding="utf-8") def write_csv(path: str, rows: Iterable[Dict[str, Any]], fieldnames: List[str]) -> None: with open(path, "w", newline="", encoding="utf-8") as fh: writer = csv.DictWriter(fh, fieldnames=fieldnames) writer.writeheader() for row in rows: writer.writerow({key: row.get(key, "") for key in fieldnames}) def write_findings_csv(analysis: JCLAnalysis, path: str) -> None: rows = [] for item in analysis.findings: row = asdict(item) row["fingerprint"] = fingerprint_finding(item) rows.append(row) write_csv(path, rows, [ "fingerprint", "severity", "rule_id", "category", "title", "message", "recommendation", "confidence", "line_no", "object_type", "object_name", "evidence", ]) def write_steps_csv(analysis: JCLAnalysis, path: str) -> None: rows = [] for step in analysis.steps: rows.append({ "name": step.name, "program": step.program, "procedure": step.procedure, "line_no": step.line_no, "dd_count": step.dd_count, "is_sort": step.is_sort, "is_utility": step.is_utility, "is_compiler": step.is_compiler, "uses_proc": step.uses_proc, "parameters": json.dumps(step.parameters, sort_keys=True), }) write_csv(path, rows, ["name", "program", "procedure", "line_no", "dd_count", "is_sort", "is_utility", "is_compiler", "uses_proc", "parameters"]) def write_dds_csv(analysis: JCLAnalysis, path: str) -> None: rows = [] for dd in analysis.dds: rows.append({ "step_name": dd.step_name, "name": dd.name, "dataset": dd.dataset, "line_no": dd.line_no, "disposition": dd.disposition, "unit": dd.unit, "space": dd.space, "dcb": dd.dcb, "is_inline": dd.is_inline, "is_sysout": dd.is_sysout, "is_dummy": dd.is_dummy, "sms_classes": json.dumps(dd.sms_classes, sort_keys=True), "raw_parameters": dd.raw_parameters, "parameters": json.dumps(dd.parameters, sort_keys=True), }) write_csv(path, rows, ["step_name", "name", "dataset", "line_no", "disposition", "unit", "space", "dcb", "is_inline", "is_sysout", "is_dummy", "sms_classes", "raw_parameters", "parameters"]) def write_datasets_csv(analysis: JCLAnalysis, path: str) -> None: write_csv(path, [asdict(item) for item in analysis.datasets], [ "dataset", "step_name", "dd_name", "disposition", "access", "is_gdg", "is_temp", "is_sensitive", "is_output", "line_no", ]) def write_category_scores_csv(analysis: JCLAnalysis, path: str) -> None: write_csv(path, [asdict(item) for item in analysis.category_scores], ["category", "finding_count", "risk_points", "highest_severity"]) def write_rules_catalog(path: str) -> None: Path(path).write_text(json.dumps(RULE_CATALOG, indent=2, sort_keys=True), encoding="utf-8") def severity_class(severity: str) -> str: return { "CRITICAL": "sev-critical", "ERROR": "sev-error", "WARNING": "sev-warning", "INFO": "sev-info", }.get(severity, "sev-info") def render_html_report(analysis: JCLAnalysis) -> str: metrics = analysis.metrics rows_findings = [] for f in sorted(analysis.findings, key=lambda item: (SEVERITY_ORDER[item.severity], item.line_no), reverse=True): rows_findings.append( f"{html.escape(f.severity)}" f"{html.escape(f.rule_id)}{html.escape(f.category)}" f"{html.escape(f.title)}{html.escape(f.message)}" f"{html.escape(f.recommendation)}{html.escape(f.confidence)}{f.line_no}" ) rows_steps = [] for s in analysis.steps: tags = [] if s.is_sort: tags.append("SORT") if s.is_utility: tags.append("UTILITY") if s.uses_proc: tags.append("PROC") rows_steps.append( f"{html.escape(s.name)}{html.escape(s.program)}{html.escape(s.procedure)}" f"{s.dd_count}{html.escape(', '.join(tags))}{s.line_no}" ) rows_dds = [] for dd in analysis.dds: rows_dds.append( f"{html.escape(dd.step_name)}{html.escape(dd.name)}{html.escape(dd.dataset)}" f"{html.escape(dd.disposition)}{html.escape(dd.space)}{html.escape(dd.unit)}{dd.line_no}" ) rows_datasets = [] for ds in analysis.datasets: rows_datasets.append( f"{html.escape(ds.dataset)}{html.escape(ds.step_name)}{html.escape(ds.dd_name)}" f"{html.escape(ds.access)}{'yes' if ds.is_gdg else ''}{'yes' if ds.is_sensitive else ''}{ds.line_no}" ) rows_categories = [] for category in analysis.category_scores: rows_categories.append( f"{html.escape(category.category)}{category.finding_count}{category.risk_points}" f"{html.escape(category.highest_severity)}" ) comparison_html = "" if analysis.comparison: cmp = analysis.comparison comparison_html = f"""

Baseline comparison

{cmp.delta_risk_score:+d}
Risk delta
{cmp.delta_finding_count:+d}
Finding delta
{len(cmp.new_rule_ids)}
New rule ids
{len(cmp.resolved_rule_ids)}
Resolved rule ids

Baseline: {html.escape(cmp.baseline_source)}

New rules: {html.escape(', '.join(cmp.new_rule_ids) or 'None')}

Resolved rules: {html.escape(', '.join(cmp.resolved_rule_ids) or 'None')}

""" return f""" JCL Doctor V2 Report

JCL Doctor V2 Report

Source: {html.escape(analysis.source_name)} | Profile: {html.escape(analysis.profile)} | Job: {html.escape(analysis.job.name)} | Status: {html.escape(analysis.status)}
{analysis.risk_score}/100
Risk score
{html.escape(analysis.highest_severity)}
Highest severity
{metrics.get('step_count')}
Steps
{metrics.get('dd_count')}
DD statements
{metrics.get('critical_count')}
Critical findings
{metrics.get('error_count')}
Errors
{metrics.get('warning_count')}
Warnings
{metrics.get('dataset_count')}
Datasets
{comparison_html}

Risk by category

{''.join(rows_categories)}
CategoryFindingsRisk pointsHighest severity

Findings

{''.join(rows_findings)}
SeverityRuleCategoryTitleMessageRecommendationConfidenceLine

Steps

{''.join(rows_steps)}
StepProgramProcedureDD countTagsLine

DD statements

{''.join(rows_dds)}
StepDDDatasetDISPSPACEUNITLine

Datasets

{''.join(rows_datasets)}
DatasetStepDDAccessGDGSensitiveLine
""" def write_html_report(analysis: JCLAnalysis, path: str) -> None: Path(path).write_text(render_html_report(analysis), encoding="utf-8") def print_summary(analysis: JCLAnalysis, max_findings: int = 14) -> None: print("JCL Doctor V2") print(f"Source : {analysis.source_name}") print(f"Profile : {analysis.profile}") print(f"Job : {analysis.job.name}") print(f"Status : {analysis.status}") print(f"Highest severity : {analysis.highest_severity}") print(f"Risk score : {analysis.risk_score}/100") print(f"Steps : {analysis.metrics['step_count']}") print(f"DD statements : {analysis.metrics['dd_count']}") print(f"Datasets : {analysis.metrics['dataset_count']}") print(f"Sensitive DSNs : {analysis.metrics['sensitive_dataset_count']}") print(f"Findings : {analysis.metrics['finding_count']}") if analysis.comparison: print(f"Risk delta : {analysis.comparison.delta_risk_score:+d}") print(f"Finding delta : {analysis.comparison.delta_finding_count:+d}") print() print("TOP FINDINGS") print("-" * 96) ordered = sorted(analysis.findings, key=lambda f: (SEVERITY_ORDER[f.severity], f.line_no), reverse=True) for finding in ordered[:max_findings]: print(f"{finding.severity:8} {finding.rule_id:28} line={finding.line_no:<4} {finding.title}") print(f" {finding.message}") def load_input(args: argparse.Namespace) -> Tuple[str, str]: if args.write_sample_jcl: sample = SAMPLES.get(args.sample_name, SAMPLE_JCL_PAYROLL) Path(args.write_sample_jcl).write_text(sample, encoding="utf-8") print(f"Sample JCL written to: {args.write_sample_jcl}") sys.exit(0) if args.write_all_samples: output_dir = Path(args.write_all_samples) output_dir.mkdir(parents=True, exist_ok=True) for name, text in SAMPLES.items(): path = output_dir / f"sample_jcl_v2_{name}.jcl" path.write_text(text, encoding="utf-8") print(f"Sample written to: {path}") sys.exit(0) if args.export_rules: write_rules_catalog(args.export_rules) print(f"Rule catalog written to: {args.export_rules}") sys.exit(0) if args.demo: return SAMPLES.get(args.demo, SAMPLE_JCL_PAYROLL), f"demo_{args.demo}" if not args.input_file: raise SystemExit("Missing input_file. Use --demo or provide a JCL file.") path = Path(args.input_file) return path.read_text(encoding="utf-8", errors="ignore"), str(path) def build_arg_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="JCL Doctor V2 - professional JCL linter and analyzer") parser.add_argument("input_file", nargs="?", help="JCL text file to analyze") parser.add_argument("--demo", choices=sorted(SAMPLES.keys()), help="Run a built-in sample JCL") parser.add_argument("--sample-name", choices=sorted(SAMPLES.keys()), default="payroll", help="Sample name for --write-sample-jcl") parser.add_argument("--write-sample-jcl", help="Write one built-in sample JCL to a file and exit") parser.add_argument("--write-all-samples", help="Write all built-in samples to a directory and exit") parser.add_argument("--profile", choices=sorted(VALID_PROFILES), default="production", help="Analysis profile") parser.add_argument("--custom-policy", help="Optional JSON policy file with sensitive_patterns and banned_patterns") parser.add_argument("--compare-with", help="Optional baseline JCL file to compare with current analysis") parser.add_argument("--json", dest="json_output", help="Write JSON analysis output") parser.add_argument("--html", dest="html_output", help="Write HTML report") parser.add_argument("--csv-findings", help="Write findings CSV") parser.add_argument("--csv-steps", help="Write steps CSV") parser.add_argument("--csv-dds", help="Write DD statements CSV") parser.add_argument("--csv-datasets", help="Write datasets CSV") parser.add_argument("--csv-categories", help="Write category scores CSV") parser.add_argument("--export-rules", help="Export the built-in rule catalog and exit") parser.add_argument("--redact", action="store_true", help="Redact dataset names in outputs") parser.add_argument("--fail-on", choices=["INFO", "WARNING", "ERROR", "CRITICAL"], help="Return exit code 2 if findings reach this severity") parser.add_argument("--max-findings-console", type=int, default=14, help="Maximum findings displayed in console summary") return parser def main(argv: Optional[List[str]] = None) -> int: parser = build_arg_parser() args = parser.parse_args(argv) text, source_name = load_input(args) policy = load_custom_policy(args.custom_policy) baseline_analysis: Optional[JCLAnalysis] = None comparison: Optional[JCLComparison] = None if args.compare_with: baseline_path = Path(args.compare_with) baseline_analysis = analyze_jcl( baseline_path.read_text(encoding="utf-8", errors="ignore"), source_name=str(baseline_path), profile=args.profile, redact=args.redact, custom_policy=policy, ) analysis = analyze_jcl(text, source_name=source_name, profile=args.profile, redact=args.redact, custom_policy=policy) if baseline_analysis: analysis.comparison = compare_analyses(analysis, baseline_analysis) print_summary(analysis, max_findings=args.max_findings_console) if args.json_output: write_json(analysis, args.json_output) if args.html_output: write_html_report(analysis, args.html_output) if args.csv_findings: write_findings_csv(analysis, args.csv_findings) if args.csv_steps: write_steps_csv(analysis, args.csv_steps) if args.csv_dds: write_dds_csv(analysis, args.csv_dds) if args.csv_datasets: write_datasets_csv(analysis, args.csv_datasets) if args.csv_categories: write_category_scores_csv(analysis, args.csv_categories) if args.fail_on and any(severity_at_least(f.severity, args.fail_on) for f in analysis.findings): return 2 return 0 if __name__ == "__main__": raise SystemExit(main())