Script python : mise à jour de token et fichier log

Résolu/Fermé
Telnaz - 12 oct. 2022 à 09:18
 Telnaz - 14 oct. 2022 à 12:44

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

A voir également:

2 réponses

mamiemando Messages postés 33459 Date d'inscription jeudi 12 mai 2005 Statut Modérateur Dernière intervention 8 janvier 2025 7 813
12 oct. 2022 à 10:48

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

1

Bonjour et merci pour votre réponse.

Je n'ai jamais utilisé asyncio ou même tout ce qui est co-routine, tâche asynchrone, etc. en python. Je me suis donc référencé pour mieux comprendre votre code.

Pour l'instant, j'ai fait le test avec une période de 5 secondes et cela fonctionne. Je pense que je vais garder finalement une période assez petite.

Pour revenir à l'API, l'outil révoque mon token toutes les 30 minutes (par défaut et par sécurité). En fait, je peux même augmenter ce temps de révocation mais je trouve que c'est plutôt bien de laisser ceci comme ça.

yg_be:

"C'est certainement plus simple, le token sera automatiquement préservé, sans rien faire de particulier."

Le programme étant toujours en cours, il faut donc que je créé une variable globale qui est modifiée à chaque régénération, tout simplement, non ?

0
mamiemando Messages postés 33459 Date d'inscription jeudi 12 mai 2005 Statut Modérateur Dernière intervention 8 janvier 2025 7 813 > Telnaz
13 oct. 2022 à 10:34

Pour revenir à l'API, l'outil révoque mon token toutes les 30 minutes (par défaut et par sécurité). En fait, je peux même augmenter ce temps de révocation mais je trouve que c'est plutôt bien de laisser ceci comme ça.

Ah ok ! En fait, cette considération dépend de la longueur de tes tokens :

  • Si on parle de tokens à la Twitter ou Github, ce sont des tokens tellement longs (plusieurs dizaines de caractères alphanumériques) que ça n'a pas vraiment d'utilité de les rafraîchir régulièrement (voire, de les rafraîchir tout court). Ce genre de token est bien adapté pour une application, car on n'a pas à les saisir manuellement. On les configure une fois pour toutes le token dans l'application qui souhaite accéder à l'API. Si ton API te propose ce genre de mécanisme, c'est à mon avis le plus propre (moins de code) tout en restant sûr.
  • Si on parle de tokens à la Microsoft Authenticator, ce sont des token beaucoup plus petits et sur un alphabet bien plus restreint (typiquement, 6 chiffres). Ils sont volontairement très court, car prévus pour être saisis manuellement par l'utilisateur. Mais du coup, ils sont aussi bien plus facile à bruteforcer, et c'est pourquoi il vaut mieux les rafraîchir régulièrement. Dans ce cas de figure, ta question à tout son sens.

Bonne chance

0
mamiemando Messages postés 33459 Date d'inscription jeudi 12 mai 2005 Statut Modérateur Dernière intervention 8 janvier 2025 7 813 > Telnaz
Modifié le 13 oct. 2022 à 11:31
Traceback (most recent call last):
  File "script.py", line 107, in <module>
    main()
  File "script.py", line 45, in main
    except asyncio.exceptions.CancelledError:
AttributeError: module 'asyncio' has no attribute 'exceptions'
  • Je n'ai pas cette erreur de mon côté. C'est sans doute lié à ta version de python (personnellement je suis en 3.10 pour le moment). Mais effectivement, quelle que soit ta version de python, tu as entièrement raison, il vaudrait mieux utiliser asyncio.CancelError (qui semble tout le temps marcher) au lieu de  asyncio.exceptions.CancelError. Merci pour cette correction ;-)

Concernant le reste de ton message #9, ta compréhension du script est correcte.

  • Faute de critère d'arrêt (par exemple un nombre de rafraîchissement) on répète indéfiniment la fonction periodic. et donc le loop.run_until_complete ne se termine jamais à moins de l'interrompre avec un signal (par exemple suite à un ctrl+C).
  • Afin de fermer proprement la boucle d'événements (et en particulier les tâches asynchrones en court), on rattrape le signal et quand il a lieu, on met fin à la coroutine periodic qui est en cours.

À moins que tu aies d'autres questions, peux-tu basculer ce sujet en résolu ?

Bonne chance

1
yg_be Messages postés 23425 Date d'inscription lundi 9 juin 2008 Statut Contributeur Dernière intervention 12 janvier 2025 1 557 > Telnaz
13 oct. 2022 à 11:54

En effet, il suffit de mettre à jour une variable (permanente), globale ou pas, suivant la façon dont le script est écrit.

1
mamiemando Messages postés 33459 Date d'inscription jeudi 12 mai 2005 Statut Modérateur Dernière intervention 8 janvier 2025 7 813 > yg_be Messages postés 23425 Date d'inscription lundi 9 juin 2008 Statut Contributeur Dernière intervention 12 janvier 2025
14 oct. 2022 à 10:48

Concernant le message #14 : pour les prochaines fois, vu qu'il s'agit d'un autre problème, même s'il te paraît mineur, ouvre une nouvelle discussion pour garder le fil de discussion propre et spécifique à un seul problème.

Réponse rapide :

Si on parle d'une méthode, tu peux aussi mémoriser l'information en question dans l'instance de l'objet que tu manipules.

WHO = "world"

class MyObjectGlob:
    def __init__(self):
        # ...
    def my_method(self, x, y, z):
        print("Hello " + WHO)
        return x + y + z

class MyObject:
    def __init__(self, who):
        self.who = who
    def my_method(self, x, y, z):
        print("Hello " + self.who)
        return x + y + z

Si on parle d'une fonction :

  • Soit tu peux ajouter cette information en tant que paramètre, c'est la solution la solution que propose yg_be (voir message #15). Cela suppose cependant de pouvoir se le permettre (car à moins d'assigner une valeur par défaut à ce nouveau paramètre, il faudra corriger en conséquence tous les appels faits à cette fonction).
    WHO = "world"
    
    def f_glob(x, y, z):
        print("Hello " + WHO)
        return x + y +z
    
    def f(x, y, z, who):
        print("Hello " + who)
        return x + y +z
  • Soit tu ne veux/peux pas modifier le paramètre de la fonction, typiquement parce que cette fonction est passée en callback d'une autre fonction pour tu ne peux pas changer le prototype. Dans ce cas, pour s'en sortir, tu peux créer un foncteur. Un foncteur est une instance d'objet qui se comporte comme une fonction : l'instance est dite callable (au même titre qu'une fonction ordinaire). On mémorise l'information à la construction du foncteur. Lorsqu'on appelle le foncteur (c'est-à-dire qu'on l'utilise comme si c'était une fonction ordinaire), c'est sa méthode __call__ qui est appelée.
    class Fonctor:
        def __init__(self, who):
            self.who = who
        def __call__(self, x, y ,z):
            print("Hello " + self.who)
            return x + y + z
    
    f = Fonctor("world")
    print(f(1, 2, 3))

Bonne chance

1
yg_be Messages postés 23425 Date d'inscription lundi 9 juin 2008 Statut Contributeur Dernière intervention 12 janvier 2025 Ambassadeur 1 557
12 oct. 2022 à 09:46

bonjour,

tout sera plus clair quand tu auras fait la boucle.

0

En fait je ne l'ai pas faite car je ne sais pas si je fais juste un Cron qui exécute le script toutes 5 ou 10 minutes ou si je fais un "While True" avec un sleep directement dans le code.

0
yg_be Messages postés 23425 Date d'inscription lundi 9 juin 2008 Statut Contributeur Dernière intervention 12 janvier 2025 1 557 > Telnaz
12 oct. 2022 à 09:53

Quand tu auras décidé cela, tu pourras commencer à réfléchir à comment préserver "token" d'une exécution à l'autre.

0
Telnaz > yg_be Messages postés 23425 Date d'inscription lundi 9 juin 2008 Statut Contributeur Dernière intervention 12 janvier 2025
12 oct. 2022 à 09:56

Ok, je pense que faire la boucle à l'intérieur du programme est mieux non ? Plutôt que d'exécuter le script par crontab à chaque fois.

0
yg_be Messages postés 23425 Date d'inscription lundi 9 juin 2008 Statut Contributeur Dernière intervention 12 janvier 2025 1 557 > Telnaz
12 oct. 2022 à 10:36

C'est certainement plus simple, le token sera automatiquement préservé, sans rien faire de particulier.

0