Rundeck Job Governance. (Look for risky patterns and highlight)

readme.md

**

Rundeck Job Governance

** Check ALL rundeck jobs for bad programming patterns. Includes by default the following for Bash and Python

• Check for Hardcoded Passwords/API tokens
• Insecure File Permissions on scripts
• Executing external commands in python
• http endpoints
• deprecated libraries or functions

(Add more with Regex)

Presently supports command, and inline scripts but can easily be modified for additional job types.

Highlights the non compliant jobs in red.

Usage. Add your Rundeck URL/API Token

Supplied as is for experimentation.No warranties. Justyn Roberts

main.py

import requests
import re
import time

base_url = "https://YOURURL"
headers = {
        'X-Rundeck-Auth-token': 'ADDTOKEN',
        'Content-Type': 'application/json',
        'accept': 'application/json',
    }



search_patterns = [
    "(?i)password\\s*=\\s*['\"](?!@option\\.[a-zA-Z_]+@).+['\"]",   # Hardcoded Passwords
    "(?i)token\\s*=\\s*['\"](?!@option\\.[a-zA-Z_]+@).+['\"]",      # Hardcoded Passwords
    "(?i)apitoken\\s*=\\s*['\"](?!@option\\.[a-zA-Z_]+@).+['\"]",   # Hardcoded Passwords
    "eval\\s*\\(.+\\)",                                             # Insecure Use of `eval`
    "subprocess\\.run\\(.+,\\s*shell=True",                         # Use of `subprocess` with `shell=True`
    # "\\$\\(\\s*[^\\)]*\\)",                                         # Potential Command Injection in Bash
    "chmod\\s+(777|666)",                                           # Insecure File Permissions
    "pickle\\.loads\\(.+\\)",                                       # Use of `pickle` in Python
    "os\\.system\\(.+\\)",                                          # Executing External Commands in Python
    #"read\\s+[a-zA-Z_]+",                                           # Unfiltered User Input in Bash
    "setenforce\\s+0|selinux=0",                                    # Disabling Security Features
    "import\\s+BaseHTTPServer|SimpleHTTPServer",                    # Use of Deprecated Libraries or Functions
    "http://"                                                       # Use of HTTP instead of HTTPS
]
keys_to_check = ["script", "exec"]  # Add more types of step here if needed


def get_projects(base_url, headers):
    url = f"{base_url}/api/11/projects"
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        return response.json()
    else:
        print(f"Error fetching projects: {response.text}")
        return []

def get_project_jobs(base_url, project_name, headers):
    url = f"{base_url}/api/14/project/{project_name}/jobs"
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        return response.json()
    else:
        print(f"Error fetching jobs for project {project_name}: {response.text}")
        return []

def get_job_workflow(base_url, job_id, headers):
    url = f"{base_url}/api/45/job/{job_id}/workflow"
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        return response.json()
    else:
        print(f"Error fetching workflow for job {job_id}: {response.text}")
        return None

def highlight_pattern(text, pattern):
    """Highlights the pattern in the text."""
    return re.sub(pattern, lambda x: f"\033[91m{x.group(0)}\033[0m", text)

def main():
    projects = get_projects(base_url, headers)
    start_time = time.time()
    total_jobs = 0
    jobs_with_issues = 0

    print("All jobs found with specified step types: "+str(keys_to_check))
    print("----------------------------------------")

    for project in projects:
        project_name = project.get('name')
        jobs = get_project_jobs(base_url, project_name, headers)

        for job in jobs:
            job_id = job.get('id')
            job_name = job.get('name')
            print(f"Project: {project_name}, Job Name: {job_name}, ID: {job_id}")
            total_jobs += 1

    print("\nPotential issues:")
    print("-----------------")
    print (search_patterns)
    print("-----------------")
    for project in projects:
        project_name = project.get('name')
        jobs = get_project_jobs(base_url, project_name, headers)

        for job in jobs:
            job_id = job.get('id')
            job_name = job.get('name')
            workflow = get_job_workflow(base_url, job_id, headers)

            if workflow:
                for step in workflow.get('workflow', []):
                    for key in keys_to_check:
                        command = step.get(key)
                        if command:
                            for pattern in search_patterns:
                                if re.search(pattern, command):
                                    highlighted_command = highlight_pattern(command, pattern)
                                    print(f"🔺 Project: {project_name}, Job Name: {job_name}, ID: {job_id}, {key.capitalize()}: {highlighted_command}")
                                    jobs_with_issues += 1
                                    break  # Break if any pattern is found

    end_time = time.time()
    total_time = end_time - start_time

    print("\nStatistics:")
    print("-----------")
    print(f"Total time to run report: {total_time:.2f} seconds")
    print(f"Total jobs scanned: {total_jobs}")
    print(f"Steps with potential issues: {jobs_with_issues}")

if __name__ == "__main__":
    main()