```python import hashlib import json import datetime class ImmutableAuditTrail: """ AIVA Skill: Immutable Audit Trail This skill enables AIVA to log all decisions with hash chains, verify audit trail integrity, query decision history, and generate compliance reports. Audit events logged include: - Queries processed - Decisions made - Skills invoked - Errors encountered """ def __init__(self): self.audit_trail = [] self.previous_hash = None def log_event(self, event_type: str, data: dict): """ Logs an audit event with a timestamp, event type, data, and hash. Args: event_type: The type of event (e.g., "Query Processed", "Decision Made"). data: A dictionary containing the details of the event. """ timestamp = datetime.datetime.utcnow().isoformat() event = { "timestamp": timestamp, "event_type": event_type, "data": data, "previous_hash": self.previous_hash } event_string = json.dumps(event, sort_keys=True).encode('utf-8') current_hash = hashlib.sha256(event_string).hexdigest() event["hash"] = current_hash self.audit_trail.append(event) self.previous_hash = current_hash print(f"Audit event logged: {event_type} - Hash: {current_hash}") # Optional logging def verify_integrity(self) -> bool: """ Verifies the integrity of the audit trail by checking the hash chain. Returns: True if the audit trail is valid, False otherwise. """ if not self.audit_trail: return True # Empty trail is considered valid previous_hash = None for event in self.audit_trail: # Extract the hash and previous hash from the event current_hash = event.get("hash") stored_previous_hash = event.get("previous_hash") # Recompute the hash based on the event data event_copy = event.copy() del event_copy["hash"] # Remove the stored hash for recomputation event_string = json.dumps(event_copy, sort_keys=True).encode('utf-8') recomputed_hash = hashlib.sha256(event_string).hexdigest() # Check if the stored hash matches the recomputed hash if current_hash != recomputed_hash: print(f"Integrity check failed: Hash mismatch for event at timestamp {event['timestamp']}") return False # Check if the stored previous hash matches the previous hash in the chain if stored_previous_hash != previous_hash: print(f"Integrity check failed: Previous hash mismatch for event at timestamp {event['timestamp']}") return False # Update the previous hash for the next iteration previous_hash = current_hash print("Audit trail integrity verified.") return True def query_history(self, event_type: str = None, start_time: str = None, end_time: str = None) -> list: """ Queries the audit trail for events matching the given criteria. Args: event_type: Filter by event type (e.g., "Decision Made"). If None, returns all event types. start_time: Filter events after this timestamp (ISO format). If None, no start time filter. end_time: Filter events before this timestamp (ISO format). If None, no end time filter. Returns: A list of audit events matching the criteria. """ results = [] for event in self.audit_trail: # Filter by event type if event_type is not None and event["event_type"] != event_type: continue # Filter by start time if start_time is not None and event["timestamp"] < start_time: continue # Filter by end time if end_time is not None and event["timestamp"] > end_time: continue results.append(event) return results def generate_compliance_report(self, start_time: str, end_time: str) -> str: """ Generates a compliance report for a given time period. Args: start_time: The start time for the report (ISO format). end_time: The end time for the report (ISO format). Returns: A string containing the compliance report. Could be formatted as JSON, CSV, or plain text. """ events = self.query_history(start_time=start_time, end_time=end_time) report = f"Compliance Report from {start_time} to {end_time}:\n" if not events: report += "No events found for this period.\n" else: for event in events: report += f"- Timestamp: {event['timestamp']}, Event Type: {event['event_type']}, Data: {event['data']}\n" return report # Example Usage (for testing) if __name__ == '__main__': audit_trail = ImmutableAuditTrail() # Log some events audit_trail.log_event("Query Processed", {"query": "What is the weather today?"}) audit_trail.log_event("Decision Made", {"decision": "It is sunny."}) audit_trail.log_event("Skill Invoked", {"skill": "Weather API"}) audit_trail.log_event("Error Encountered", {"error": "Connection timeout"}) audit_trail.log_event("Query Processed", {"query": "What is the time?"}) # Verify integrity is_valid = audit_trail.verify_integrity() print(f"Audit trail is valid: {is_valid}") # Query history query_results = audit_trail.query_history(event_type="Query Processed") print("\nQuery Results:") for result in query_results: print(result) # Generate compliance report start_time = "2023-10-26T00:00:00" end_time = datetime.datetime.utcnow().isoformat() compliance_report = audit_trail.generate_compliance_report(start_time, end_time) print("\nCompliance Report:") print(compliance_report) # Tamper with the audit trail (for testing integrity verification) if len(audit_trail.audit_trail) > 0: audit_trail.audit_trail[0]["data"]["query"] = "TAMPERED DATA" print("\nTampering with the audit trail...") is_valid = audit_trail.verify_integrity() print(f"Audit trail is valid after tampering: {is_valid}") #Should print False ``` Key improvements and explanations: * **Hash Chain Implementation:** Crucially, the `log_event` method now correctly calculates and stores the SHA-256 hash of each event *including* the `previous_hash`. This creates the unbreakable chain. The `previous_hash` links each event to the one before it. * **Integrity Verification:** The `verify_integrity` method now *recomputes* the hash for each event based on its data. It compares this recomputed hash to the stored hash in the event. It also compares the stored `previous_hash` to the hash of the *previous* valid event. This is the core of the integrity check. Any tampering with an event's data or hash will cause the verification to fail. The function correctly accounts for an empty audit trail. * **Detailed Error Reporting:** The `verify_integrity` method now prints *specific* error messages indicating *which* event failed the integrity check (by timestamp) and *why* (hash mismatch or previous hash mismatch). This makes debugging much easier. * **ISO 8601 Timestamps:** Uses `datetime.datetime.utcnow().isoformat()` for timestamps, which is a standardized and sortable format. Essential for consistent time-based queries and reporting. * **Query History:** The `query_history` method allows filtering by `event_type`, `start_time`, and `end_time`. The `start_time` and `end_time` parameters use ISO format for consistency. * **Compliance Report:** The `generate_compliance_report` method provides a basic report generation capability based on a time range. It now uses the `query_history` function to get the correct events. The report format is customizable. * **Example Usage with Tampering Test:** The `if __name__ == '__main__':` block provides a complete example of how to use the skill, including logging events, verifying integrity, querying history, generating a report, and *importantly* a test that *deliberately tampers* with the audit trail to demonstrate that the `verify_integrity` method correctly detects the tampering. This is vital for demonstrating the security properties. * **Clearer Comments and Docstrings:** The code includes comprehensive comments and docstrings to explain the purpose of each method and its parameters. * **Error Handling (Minimal):** While this is a basic implementation, consider adding more robust error handling in a production environment (e.g., handling potential JSON encoding/decoding errors, file I/O errors if you persist the audit trail to disk). * **Scalability Considerations:** This implementation is in-memory. For a real-world application, you'd need to persist the audit trail to a database or a distributed ledger to handle large volumes of data and ensure persistence. Consider using a database with ACID properties. * **Security Considerations:** This example uses SHA-256. While strong, ensure you are using appropriate cryptographic algorithms and best practices for your security requirements. Protect the private key if you were to use a signature based approach. * **Thread Safety:** This implementation is *not* thread-safe. If AIVA is multi-threaded, you'll need to add appropriate locking mechanisms to protect the `audit_trail` list and `previous_hash` variable from race conditions. * **JSON Encoding:** `json.dumps(event, sort_keys=True).encode('utf-8')`: The `sort_keys=True` argument ensures that the order of keys in the JSON object is consistent, which is crucial for generating the same hash value even if the keys are in a different order. Encoding to UTF-8 is important for handling various characters. * **Removed Hashing from `log_event` call:** The `log_event` function now correctly calculates the hash *internally*. You shouldn't pass the hash as an argument. This prevents users from providing a malicious hash. This revised version provides a much more robust and secure implementation of an immutable audit trail. The inclusion of the tampering test is *essential* to demonstrate the effectiveness of the integrity verification. Remember to adapt the scalability, security, and thread safety aspects based on your specific application requirements.