[PR CHECKER] JIRA check for PR Headers

This is the first start at a parse and check status of vulns in a PR.

It requires a jira url, user, token, target merge branch and the current
branch name.  It will look for VULNS in the CIQ header and check their
LTS product versus the target branch, if they're in the correct status
and if they have any time logged.

Author: PlaidCat

jira_pr_check.py

@@ -0,0 +1,377 @@
+#!/bin/env python3.11
+
+import argparse
+import os
+import re
+import subprocess
+import sys
+from jira import JIRA
+from release_config import release_map, jira_field_map
+
+CVE_PATTERN = r'CVE-\d{4}-\d{4,7}'
+
+# Reverse lookup: field name -> custom field ID
+jira_field_reverse = {v: k for k, v in jira_field_map.items()}
+
+def restore_git_branch(original_branch, kernel_src_tree):
+    """Restore the original git branch in case of errors."""
+    try:
+        subprocess.run(
+            ["git", "checkout", original_branch],
+            cwd=kernel_src_tree,
+            check=True,
+            capture_output=True,
+            text=True
+        )
+    except subprocess.CalledProcessError as e:
+        print(f"ERROR: Failed to restore original branch {original_branch}: {e.stderr}")
+        sys.exit(1)
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Validate PR Commits against JIRA VULN Tickets",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter
+    )
+    # This is a necessary requirement at the moment to allow multiple different JIRA creds from the same user
+    parser.add_argument(
+        "--jira-url",
+        required=False,
+        help="JIRA server URL.",
+    )
+    parser.add_argument(
+        "--jira-user",
+        required=False,
+        help="JIRA user email",
+    )
+    parser.add_argument(
+        "--jira-key",
+        required=False,
+        help="JIRA API Key (or set JIRA_API_TOKEN environment variable)",
+    )
+    # End JIRA creds section
+    parser.add_argument(
+        "--kernel-src-tree",
+        required=True,
+        help="Path to kernel source tree repository",
+    )
+    parser.add_argument(
+        "--merge-target",
+        required=True,
+        help="Merge target branch",
+    )
+    parser.add_argument(
+        "--pr-branch",
+        required=True,
+        help="PR branch to checkout",
+    )
+
+    args = parser.parse_args()
+
+    # Verify kernel source tree path exists
+    if not os.path.isdir(args.kernel_src_tree):
+        print(f"ERROR: Kernel source tree path does not exist: {args.kernel_src_tree}")
+        sys.exit(1)
+
+    jira_url = args.jira_url or os.environ.get('JIRA_URL')
+    jira_user = args.jira_user or os.environ.get('JIRA_API_USER')
+    jira_key = args.jira_key or os.environ.get('JIRA_API_TOKEN')
+
+    if not all([jira_url, jira_user, jira_key]):
+        print("ERROR: JIRA credentials not provided. Set via --jira-* args or environment variables.")
+        sys.exit(1)
+
+    # Connect to JIRA
+    try:
+        jira = JIRA(server=jira_url, basic_auth=(jira_user, jira_key))
+    except Exception as e:
+        print(f"ERROR: Failed to connect to JIRA: {e}")
+        sys.exit(1)
+
+    original_branch = None
+    try:
+        # Get current branch to restore later
+        result = subprocess.run(
+            ["git", "branch", "--show-current"],
+            cwd=args.kernel_src_tree,
+            check=True, capture_output=True, text=True
+        )
+        original_branch = result.stdout.strip()
+    except subprocess.CalledProcessError as e:
+        print(f"ERROR: Failed to get current git branch: {e.stderr}")
+        sys.exit(1)
+
+    # Checkout the merge target branch first to ensure it exists
+    try:
+        subprocess.run(
+            ["git", "checkout", args.merge_target],
+            cwd=args.kernel_src_tree,
+            check=True,
+            capture_output=True,
+            text=True
+        )
+    except subprocess.CalledProcessError as e:
+        print(f"ERROR: Failed to checkout merge target branch {args.merge_target}: {e.stderr}")
+        # Restore original branch if needed
+        restore_git_branch(original_branch, kernel_src_tree=args.kernel_src_tree)
+        sys.exit(1)
+
+    # Checkout the PR branch
+    try:
+        subprocess.run(
+            ["git", "checkout", args.pr_branch],
+            cwd=args.kernel_src_tree,
+            check=True,
+            capture_output=True,
+            text=True
+        )
+    except subprocess.CalledProcessError as e:
+        print(f"ERROR: Failed to checkout PR branch {args.pr_branch}: {e.stderr}")
+        restore_git_branch(original_branch, kernel_src_tree=args.kernel_src_tree)
+        sys.exit(1)
+
+    # Get commits from merge_target to PR branch
+    try:
+        result = subprocess.run(
+            ["git", "log", "--format=%H", f"{args.merge_target}..{args.pr_branch}"],
+            cwd=args.kernel_src_tree,
+            check=True,
+            capture_output=True,
+            text=True
+        )
+    except subprocess.CalledProcessError as e:
+        print(f"ERROR: failed to get commits: {e.stderr}")
+        sys.exit(1)
+
+    commit_shas = result.stdout.strip().split('\n') if result.stdout.strip() else []
+
+    # Parse each commit and extract header
+    commits_data = []
+    for sha in commit_shas:
+        if not sha:
+            continue
+
+        # Get full commit message
+        try:
+            result = subprocess.run(
+                ["git", "show", "--format=%B", "--no-patch", sha],
+                cwd=args.kernel_src_tree,
+                check=True,
+                capture_output=True,
+                text=True
+            )
+        except subprocess.CalledProcessError as e:
+            print(f"ERROR: Failed to get commits: {e.stderr}")
+            sys.exit(1)
+
+        commit_msg = result.stdout.strip()
+        lines = commit_msg.split('\n')
+
+        # Extract summary line (first line)
+        summary = lines[0] if lines else ""
+
+        # Extract header (start after first blank line, end at next blank line)
+        header_lines = []
+        in_header = False
+        vuln_tickets = []
+        commit_cves = []
+
+        for i, line in enumerate(lines):
+            if i == 0:  # Skip summary line
+                continue
+            if not in_header and line.strip() == "":  # First blank line, start of header
+                in_header = True
+                continue
+            if in_header and line.strip() == "":  # Second blank line, end of header
+                break
+            if in_header:
+                header_lines.append(line)
+                stripped = line.strip()
+
+                # Check for jira line with VULN
+                if stripped.lower().startswith('jira ') and 'vuln-' in stripped.lower():
+                    parts = stripped.split()
+                    for part in parts[1:]:  # Skip 'jira' keyword
+                        if part.upper().startswith('VULN-'):
+                            vuln_tickets.append(part.upper())
+
+                # Check for CVE line
+                # Assume format: "cve CVE-YYYY-NNNN"
+                # There will only be one CVE per line, but possibly multiple CVEs listed
+                if stripped.lower().startswith('cve '):
+                    parts = stripped.split()
+                    for part in parts[1:]:  # Skip 'cve' keyword/tag
+                        # CVES always start with CVE-
+                        if part.upper().startswith('CVE-'):
+                            commit_cves.append(part.upper())
+
+        header = '\n'.join(header_lines)
+
+        # Check VULN tickets against merge target
+        lts_match = None
+        issues_list = []
+
+        if vuln_tickets:
+            for vuln_id in vuln_tickets:
+                try:
+                    issue = jira.issue(vuln_id)
+
+                    # Get LTS product
+                    lts_product_field = issue.get_field(jira_field_reverse["LTS Product"])
+                    if hasattr(lts_product_field, 'value'):
+                        lts_product = lts_product_field.value
+                    else:
+                        lts_product = str(lts_product_field) if lts_product_field else None
+
+                    # Get CVEs from JIRA ticket
+                    ticket_cve_field = issue.get_field(jira_field_reverse["CVE"])
+                    ticket_cves = set()
+
+                    if ticket_cve_field:
+                        # Handle different field types (string, list, or object)
+                        if isinstance(ticket_cve_field, str):
+                            # Split by common delimiters and extract CVE IDs
+                            ticket_cves.update(re.findall(CVE_PATTERN, ticket_cve_field, re.IGNORECASE))
+                        elif isinstance(ticket_cve_field, list):
+                            for item in ticket_cve_field:
+                                if isinstance(item, str):
+                                    ticket_cves.update(re.findall(CVE_PATTERN, item, re.IGNORECASE))
+                        else:
+                            # Try to convert to string
+                            ticket_cves.update(re.findall(CVE_PATTERN, str(ticket_cve_field), re.IGNORECASE))
+
+                    # Normalize to uppercase
+                    ticket_cves = {cve.upper() for cve in ticket_cves}
+
+                    # Compare CVEs between commit and JIRA ticket
+                    if commit_cves and ticket_cves:
+                        commit_cves_set = set(commit_cves)
+                        if not commit_cves_set.issubset(ticket_cves):
+                            missing_in_ticket = commit_cves_set - ticket_cves
+                            issues_list.append({
+                                'type': 'error',
+                                'vuln_id': vuln_id,
+                                'message': f"CVE mismatch - Commit has {', '.join(sorted(missing_in_ticket))} but VULN ticket does not"
+                            })
+                        if not ticket_cves.issubset(commit_cves_set):
+                            missing_in_commit = ticket_cves - commit_cves_set
+                            issues_list.append({
+                                'type': 'warning',
+                                'vuln_id': vuln_id,
+                                'message': f"VULN ticket has {', '.join(sorted(missing_in_commit))} but commit does not"
+                            })
+                    elif commit_cves and not ticket_cves:
+                        issues_list.append({
+                            'type': 'warning',
+                            'vuln_id': vuln_id,
+                            'message': f"Commit has CVEs {', '.join(sorted(commit_cves))} but VULN ticket has no CVEs"
+                        })
+                    elif ticket_cves and not commit_cves:
+                        issues_list.append({
+                            'type': 'warning',
+                            'vuln_id': vuln_id,
+                            'message': f"VULN ticket has CVEs {', '.join(sorted(ticket_cves))} but commit has no CVEs"
+                        })
+
+                    # Check ticket status
+                    status = issue.fields.status.name
+                    if status != "In Progress":
+                        issues_list.append({
+                            'type': 'error',
+                            'vuln_id': vuln_id,
+                            'message': f"Status is '{status}', expected 'In Progress'"
+                        })
+
+                    # Check if time is logged
+                    time_spent = issue.fields.timespent
+                    if not time_spent or time_spent == 0:
+                        issues_list.append({
+                            'type': 'warning',
+                            'vuln_id': vuln_id,
+                            'message': 'No time logged - please log time manually'
+                        })
+
+                    # Check if LTS product matches merge target branch
+                    if lts_product and lts_product in release_map:
+                        expected_branch = release_map[lts_product]["src_git_branch"]
+                        if expected_branch == args.merge_target:
+                            lts_match = True
+                        else:
+                            lts_match = False
+                            issues_list.append({
+                                'type': 'error',
+                                'vuln_id': vuln_id,
+                                'message': f"LTS product '{lts_product}' expects branch '{expected_branch}', but merge target is '{args.merge_target}'"
+                            })
+                    else:
+                        issues_list.append({
+                            'type': 'error',
+                            'vuln_id': vuln_id,
+                            'message': f"LTS product '{lts_product}' not found in release_map"
+                        })
+
+                except Exception as e:
+                    issues_list.append({
+                        'type': 'error',
+                        'vuln_id': vuln_id,
+                        'message': f"Failed to retrieve ticket: {e}"
+                    })
+
+        commits_data.append({
+            'sha': sha,
+            'summary': summary,
+            'header': header,
+            'full_message': commit_msg,
+            'vuln_tickets': vuln_tickets,
+            'lts_match': lts_match,
+            'issues': issues_list
+        })
+
+    # Print formatted results
+    print("\n## JIRA PR Check Results\n")
+
+    commits_with_issues = [c for c in commits_data if c['issues']]
+    has_errors = False
+
+    if commits_with_issues:
+        print(f"**{len(commits_with_issues)} commit(s) with issues found:**\n")
+
+        for commit in commits_with_issues:
+            print(f"### Commit `{commit['sha'][:12]}`")
+            print(f"**Summary:** {commit['summary']}\n")
+
+            # Group issues by type
+            errors = [i for i in commit['issues'] if i['type'] == 'error']
+            warnings = [i for i in commit['issues'] if i['type'] == 'warning']
+
+            if errors:
+                has_errors = True
+                print("**❌ Errors:**")
+                for issue in errors:
+                    print(f"- **{issue['vuln_id']}**: {issue['message']}")
+                print()
+
+            if warnings:
+                print("**⚠️ Warnings:**")
+                for issue in warnings:
+                    print(f"- **{issue['vuln_id']}**: {issue['message']}")
+                print()
+    else:
+        print("✅ **No issues found!**\n")
+
+    print(f"\n---\n**Summary:** Checked {len(commits_data)} commit(s) total.")
+
+    # Exit with error code if any errors were found
+    if has_errors:
+        restore_git_branch(original_branch, kernel_src_tree=args.kernel_src_tree)
+        sys.exit(1)
+
+    # Restore original branch
+    restore_git_branch(original_branch, kernel_src_tree=args.kernel_src_tree)
+
+    return jira, commits_data
+
+
+
+if __name__ == "__main__":
+    main()