Script python : mise à jour de token et fichier log Résolu

Question

Bonjour,

Je viens de développer un petit script python qui va permettre de récupérer le status d'agent HIDS (Host Intrusion Detection System), via une API, et d'envoyer les données à Elasticsearch/Kibana (moteur de recherche et outil de visualisation pour ceux qui connaissent).

L'idée est que ce script soit exécuté en boucle en permanence ou toutes les 5 minutes, quelque chose dans le genre. (je n'ai pas encore fait de boucle pour l'instant)

Ensuite, il faut qu'il vérifie si le token de l'API est encore bon ou pas, sinon, j'en régénère un. Cependant, je n'ai pas réussi à imaginer la chose afin d'éviter que le token soit régénéré à chaque fois. En fait, je m'étais dit je créé une variable globale, qui est modifiée si le token est expiré, mais le raisonnement ne fonctionne pas non plus.

Pour l'instant, j'ai laissé la variable token égale à xxxx, elle est donc comparée à chaque fois afin de régénérer un token. Je voulais que cette variable prenne la valeur de la nouvelle, mais je ne vois pas comment faire ...

Lorsque j'ai le bon token, je créé un fichier log avec "logging" ou j'écris les différentes données ...

Je viens donc ici pour savoir comment faire pour le token et également pour les problèmes éventuels liés à mon code, j'aimerais savoir quels sont les soucis (notamment d'optimisation), étant débutant.

A savoir qu'il fonctionne, j'ai bien mes lignes de log dans le fichier correspondant.

Voici le code :

#!/usr/bin/python
# -*- coding: utf-8 -*-

import json
import requests
import urllib3
import logging

logging.basicConfig(filename="/var/log/test.log", level=logging.INFO,format="%(asctime)s %(message)s")

# Disable insecure https warnings (for self-signed SSL certificates)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

# global varibales
global token
token = "xxxxxx"

#### request to list data ####
def ListAgentStatus(token):

# api-endpoint
        URL1 = "https://localhost/agts"

# headers
        headers = {'Content-Type': 'application/json','Authorization': f'Bearer {token}'}

# sending get request and saving the response as response object
        response1 = requests.get(URL1, headers=headers, verify=False)

# extracting data in json format
        data = response1.json()

if response1.status_code == 200:
                # count number of data
                count = len(data['data']['affected_items'])

for i in range(count):
                        status = data['data']['affected_items'][i]['stats']

if status == "1":
                                logging.info("status: 1")
                        elif status == "2":
                                logging.critical("status: 2")
                        elif status == "3":
                                logging.warning("status: 3")
                        else:
                                logging.error("status not available")

# verification if token is expired
        elif data['title'] == "Unauthorized":
                new_token = TokenGenerate()
                ListStatus(new_token)
        else:
                raise Exception(f"Error obtaining response: {data}")

def TokenGenerate():

# api-endpoint
        URL2 = "https://localhost/security/"

# params
        params = {'raw': 'true'}

# sending get request and saving the response as response object
        response2 = requests.get(URL2, params=params, verify=False, auth=('xxxxx', 'xxxxxx'))

# conversion in text format
        new_token = response2.text

return new_token

# extracting client status

ListStatus(token)

Merci par avance

yg_be · Answer

bonjour,

tout sera plus clair quand tu auras fait la boucle.

mamiemando · Answer

Bonjour,

Pas besoin de cron, tu peux programmer une tâche qui va se déclencher toutes les 5 minutes en python avec asyncio.

Voici un exemple qui montre comment faire :

import asyncio
from signal import SIGINT, SIGTERM

PERIOD = 1 # In seconds

async def periodic(period):
    while True:
        print("periodic")
        await asyncio.sleep(period)

def exit_loop():
    for task in asyncio.all_tasks():
        #print(f"cancel {task}")
        task.cancel()

def main():
    loop = asyncio.new_event_loop()
    for signal in [SIGINT, SIGTERM]:
        loop.add_signal_handler(signal, exit_loop)
    try:
        loop.run_until_complete(periodic(PERIOD))
    except asyncio.exceptions.CancelledError:
        print("all tasks cancelled")

if __name__ == "__main__":
    main()

Après, ça me paraît un peu étrange de régénérer un token à chaque fois, mais je ne sais pas comment marche l'API dont tu parles. Si tu prends des API classiques (par exemple twitter), tu génères un token par application une fois pour toute, l'application charge ce token, et ainsi depuis ton twitter tu peux ou non révoquer l'accès de cette application. Bref, ça vaudrait le coup de voir si tu fais vraiment les choses dans les règles de l'art...

Bonne chance

Script python : mise à jour de token et fichier log

2 réponses

Discussions similaires

Newsletters