Présentation du Script query_rundeck_pwd_auth.py
Ce script Python utilise la bibliothèque requests pour interagir avec l’API REST de Rundeck en suivant l’approche suivante :
- Authentification par Mot de Passe : Le script gère l’authentification en envoyant les identifiants (
j_usernameetj_password) au point de terminaison/j_security_checkde Rundeck. Cela établit une session, et le mot de passe est saisi de manière sécurisée (non affiché) grâce au modulegetpass. - Collecte des Données :
- Il commence par lister tous les projets disponibles.
- Pour chaque projet, il liste les jobs.
- Pour chaque job, une requête API est faite pour récupérer la dernière exécution en utilisant le paramètre
max: 1. Le script parse le statut, les dates de début et de fin de cette dernière exécution.
- Sortie Standard et Exportation CSV :
- Les données sont affichées dans la console pour une vérification immédiate.
- Un fichier
rundeck_jobs.csvest généré, fournissant un enregistrement structuré et facilement utilisable pour l’analyse ou l’audit. Le format de sortie est en semicolon-separated values (CSV) :EXPORT_DATE;PROJET;JOB_ID;JOB_NAME;DERNIERE_EXECUTION.
Utilisation
L’exécution du script se fait en fournissant l’URL du site Rundeck, l’URL de l’API et le nom d’utilisateur comme arguments :
python query_rundeck_pwd_auth.py site_url api_url user
python query_rundeck_pwd_auth.py ‘http://myrundeck.mydomain:4440’ ‘http://myrundeck.mydomain:4440/api/14’ myaccount
import requests
import json
from contextlib import redirect_stdout
import sys
import getpass
import datetime
if len(sys.argv) < 4:
print('Usage: query_rundeck_pwd_auth.py <site_url> <api url> <user>')
sys.exit()
objdate = datetime.datetime
# site url
rundeck_site_url = (sys.argv[1])
# api url
rundeck_api_url = (sys.argv[2])
# user
user = (sys.argv[3])
# password demandé de façon sécurisée (non affiché)
password = getpass.getpass(prompt='Mot de passe: ')
# Open session with user/password authentication
url = f'{rundeck_site_url}/j_security_check'
session = requests.session()
response = session.post(url, data={"j_username": user, "j_password": password})
# debug minimal : afficher erreur si auth KO
if response.status_code != 200:
print(f"Auth error: {response.status_code}")
try:
print(response.text)
except Exception:
pass
# Fonction pour lister les projets
def lister_projects():
url = f'{rundeck_api_url}/projects'
headers = {
'Accept': 'application/json'
}
response = session.get(url,headers = headers)
allprojects=[]
myproject={
'project': None
}
if response.status_code == 200:
projects = response.json()
for project in projects:
project_name=project['name']
myproject={
'project_name': project_name,
}
allprojects.append(myproject)
return allprojects
else:
print(f"Erreur: {response.status_code}")
try:
print(response.text)
except Exception:
pass
# Fonction pour lister les jobs d'un projet
def lister_jobs(projet):
url = f'{rundeck_api_url}/project/{projet}/jobs'
headers = {
'Accept': 'application/json'
}
response = session.get(url,headers = headers)
alljobs=[]
myjob={
'export_date': None,
'projet': None,
'job_id': None,
'job_name': None,
'last_result': None
}
if response.status_code == 200:
jobs = response.json()
for job in jobs:
job_id=job['id']
job_name=job['name']
last_result=dernier_etat_execution(job['id'])
myjob={
'export_date': objdate.now().strftime("%d/%m/%Y"),
'projet': projet,
'job_id': job_id,
'job_name': job_name,
'last_result': last_result
}
alljobs.append(myjob)
return alljobs
else:
print(f"Erreur: {response.status_code}")
try:
print(response.text)
except Exception:
pass
# Fonction pour obtenir le dernier état d'exécution d'un job
def dernier_etat_execution(job_id):
url = f'{rundeck_api_url}/job/{job_id}/executions'
headers = {
'Accept': 'application/json'
}
params = {
'max': 1, # Limite à une exécution pour obtenir la plus récente
'offset': 0
}
response = session.get(url, headers=headers, params=params)
allexec=[]
myexec={
'id': None,
'statut': None,
'start': None,
'end': None
}
if response.status_code == 200:
executions = response.json()
if executions['executions']:
derniere_execution = executions['executions'][0]
exec_id=derniere_execution['id']
exec_statut=derniere_execution['status']
exec_start=objdate.fromisoformat(derniere_execution['date-started']['date'].replace('Z','')).strftime("%d/%m/%Y, %H:%M:%S")
# Si 'date-started' n'est pas trouvé dans la derniere execution on positionne exec_end='null'
if 'date-ended' in derniere_execution:
exec_end=objdate.fromisoformat(derniere_execution['date-ended']['date'].replace('Z','')).strftime("%d/%m/%Y, %H:%M:%S")
else:
exec_end='null'
myexec={
'id': exec_id,
'statut': exec_statut,
'start': exec_start,
'end': exec_end
}
allexec.append(myexec)
return allexec
else:
return("Aucune exécution trouvée pour ce job.")
else:
return(f"Erreur: {response.status_code}")
# All projects
allproj=lister_projects()
# Transforme la liste allproj en dictionnaire
dicallproj=dict(enumerate(allproj))
# Pour chacun des projets, recuperation de la liste des jobs et de l'état de la derniere execution
dicallprojjobs={}
for i in sorted(dicallproj.keys()):
dicallprojjobs[i]=lister_jobs(dicallproj[i]['project_name'])
# Affichage du dictionnaire imbriqué
print("ALL PROJETS - ALL LAST JOBS:")
print("EXPORT_DATE;PROJET;JOB_ID;JOB_NAME;DERNIERE_EXECUTION")
for i in sorted(dicallprojjobs.items()):
for x in i[1]:
print("{export_date};{projet};{job_id};{job_name};{last_result}".format(export_date=x['export_date'],projet=x['projet'],job_id=x['job_id'], job_name=x['job_name'], last_result=x['last_result']))
# Export vers rundeck_jobs.csv du dictionnaire dicallprojjobs
with open('rundeck_jobs.csv', 'w') as f:
with redirect_stdout(f):
print("EXPORT_DATE;PROJET;JOB_ID;JOB_NAME;DERNIERE_EXECUTION")
for i in sorted(dicallprojjobs.items()):
for x in i[1]:
print("{export_date};{projet};{job_id};{job_name};{last_result}".format(export_date=x['export_date'],projet=x['projet'],job_id=x['job_id'], job_name=x['job_name'], last_result=x['last_result']))
print("\nresultat exporté dans rundeck_jobs.csv")