1. import time
  2. import hashlib
  4. def guard_api_payload(payload, monitoring_key, monitoring_interval=5):
  5. """
  6. Guards API payload execution for monitoring purposes.
  8. Args:
  9. payload (dict): The API payload to be executed.
  10. monitoring_key (str): A unique key for monitoring the payload.
  11. monitoring_interval (int): The interval (in seconds) for monitoring.
  13. Returns:
  14. bool: True if the payload is executed, False otherwise. Returns False if monitoring fails.
  15. """
  16. try:
  17. # Generate a hash of the payload for monitoring
  18. payload_hash = hashlib.sha256(str(payload).encode('utf-8')).hexdigest()
  20. # Check if the payload has been monitored recently
  21. timestamp = time.time()
  23. # Retrieve the last monitored payload hash and timestamp (simulated storage)
  24. last_monitored_hash = get_last_monitored_hash(monitoring_key)
  25. last_monitored_timestamp = get_last_monitored_timestamp(monitoring_key)
  27. # Check if the payload is new or if it's older than the monitoring interval
  28. if payload_hash != last_monitored_hash or (timestamp - last_monitored_timestamp) > monitoring_interval:
  29. # Payload is new or hasn't been monitored recently
  30. log_payload(payload, monitoring_key) # Log the payload for monitoring
  31. execute_api_payload(payload) # Execute the API payload
  32. # Update the last monitored hash and timestamp
  33. update_last_monitored(monitoring_key, payload_hash, timestamp)
  34. return True # Execution successful
  35. else:
  36. # Payload has been monitored recently, prevent execution
  37. log_ignored_payload(payload, monitoring_key) # Log that the payload was ignored
  38. return False # Execution blocked
  39. except Exception as e:
  40. log_error(f"Error during payload guarding: {e}")
  41. return False # Indicate failure
  45. def get_last_monitored_hash(key):
  46. """Simulates retrieving the last monitored hash from a storage."""
  47. # Replace with your actual storage mechanism (e.g., database, file)
  48. try:
  49. with open("payload_monitoring_data.txt", "r") as f:
  50. lines = f.readlines()
  51. if lines:
  52. return lines[0].strip()
  53. else:
  54. return None
  55. except FileNotFoundError:
  56. return None
  58. def get_last_monitored_timestamp(key):
  59. """Simulates retrieving the last monitored timestamp from a storage."""
  60. # Replace with your actual storage mechanism
  61. try:
  62. with open("payload_monitoring_data.txt", "r") as f:
  63. lines = f.readlines()
  64. if lines:
  65. return lines[1].strip()
  66. else:
  67. return 0
  68. except FileNotFoundError:
  69. return 0
  71. def update_last_monitored(key, hash_value, timestamp):
  72. """Simulates updating the last monitored hash and timestamp in storage."""
  73. # Replace with your actual storage mechanism
  74. with open("payload_monitoring_data.txt", "w") as f:
  75. f.write(hash_value + "\n")
  76. f.write(str(timestamp) + "\n")
  79. def log_payload(payload, key):
  80. """Logs the payload for monitoring purposes."""
  81. print(f"Monitoring Key: {key}, Payload: {payload}")
  83. def log_ignored_payload(payload, key):
  84. """Logs the ignored payload for monitoring purposes."""
  85. print(f"Monitoring Key: {key}, Payload: {payload} - Ignored (already monitored)")
  87. def log_error(message):
  88. """Logs error messages."""
  89. print(f"Error: {message}")
  91. def execute_api_payload(payload):
  92. """Simulates executing the API payload."""
  93. print(f"Executing API payload: {payload}")
  95. if __name__ == '__main__':
  96. # Example Usage
  97. monitoring_key = "my_api_key"
  99. # First execution - payload is monitored
  100. payload1 = {"data": "some data"}
  101. guard_api_payload(payload1, monitoring_key)
  103. # Second execution - payload is blocked
  104. payload2 = {"data": "different data"}
  105. guard_api_payload(payload2, monitoring_key

Add your comment