Script python : mise à jour de token et fichier log
RésoluTelnaz -
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
2 réponses
-
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
-
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 ?
-
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
-
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
-
-
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 + zSi 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
- 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).
-
-
yg_be Messages postés 23437 Date d'inscription Statut Contributeur Dernière intervention Ambassadeur 1 588
bonjour,
tout sera plus clair quand tu auras fait la boucle.