Invario/Inventarsystem
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 8 Wiki Activity

implementet encryption and decryption to the audit to provide a complete lak of context for a petential attacker

Browse Source
This commit is contained in:
AIIrondev
2026-06-27 13:45:21 +02:00
parent 7e258e07ab
commit 7bedfc1558
2 changed files with 58 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Web/app.py
+22 -39
View File
@@ -388,6 +388,8 @@ ALLOWED_COVER_DOMAINS = {
"www.googleapis.com"
}
SENSITIVE_AUDIT_FIELDS = ["email", "username", "full_name", "phone"]
# Apply the configuration for general use throughout the app
APP_VERSION = __version__
RELEASE_STATE_FILE = os.path.join(os.path.dirname(BASE_DIR), '.release-version')
@@ -726,9 +728,9 @@ def _append_audit_event_standalone(event_type, payload):
al.append_audit_event(
db=db,
event_type=event_type,
actor=encrypt_text(session.get('username', 'system')),
payload=encrypt_text(str(payload)),
request_ip=encrypt_text(request.remote_addr),
actor=session.get('username', 'system'),
payload=str(payload),
request_ip=request.remote_addr,
source='web',
)
except Exception as exc:
@@ -8044,23 +8046,14 @@ def admin_audit_dashboard():
al.ensure_audit_indexes(db)
verify_result = al.verify_audit_chain(db)
audit_rows = list(
db['audit_log'].find(
{},
{
'chain_index': 1,
'event_type': 1,
'actor': 1,
'source': 1,
'ip': 1,
'timestamp': 1,
'created_at': 1,
'entry_hash': 1,
'prev_hash': 1,
'payload': 1,
}
).sort('chain_index', -1).limit(200)
)
audit_rows = list(db['audit_log'].find(...).sort('chain_index', -1).limit(200))
# DEC_START: Decrypt the sensitive fields for display
for row in audit_rows:
if "payload" in row:
# decrypt_document_fields acts in-place
decrypt_document_fields(row["payload"], SENSITIVE_AUDIT_FIELDS)
# DEC_END
return render_template(
'admin_audit.html',
@@ -8105,28 +8098,18 @@ def admin_audit_export_pdf_official():
])
)
audit_rows = list(
db['audit_log'].find(
{},
{
'chain_index': 1,
'event_type': 1,
'actor': 1,
'source': 1,
'ip': 1,
'timestamp': 1,
'created_at': 1,
'entry_hash': 1,
'prev_hash': 1,
'payload': 1,
}
).sort('chain_index', -1).limit(limit)
)
audit_rows = list(db['audit_log'].find(...).sort('chain_index', -1).limit(limit))
# DEC_START: Decrypt sensitive fields for the PDF report
for row in audit_rows:
if "payload" in row:
decrypt_document_fields(row["payload"], SENSITIVE_AUDIT_FIELDS)
# DEC_END
# Get school information from settings or use defaults
school_info = _get_school_info_for_export()
# Generate PDF
# Generate PDF with now-decrypted audit_rows
pdf_content = pdf_export.generate_audit_pdf(
verify_result=verify_result,
event_counts=event_counts,
@@ -8134,7 +8117,7 @@ def admin_audit_export_pdf_official():
export_type='official',
school_info=school_info
)
response = make_response(pdf_content)
response.headers['Content-Type'] = 'application/pdf'
response.headers['Content-Disposition'] = f'attachment; filename=audit-official-report-{datetime.datetime.utcnow().strftime("%Y%m%d-%H%M%S")}.pdf'
Web/modules/log/audit_log.py
+36 -13
View File
@@ -10,6 +10,7 @@ import hashlib
import json
import random
import time
from Web.modules.inventarsystem.data_protection import encrypt_document_fields, decrypt_document_fields
from pymongo.errors import DuplicateKeyError
@@ -24,21 +25,34 @@ def _entry_hash(prev_hash, payload):
base = f"{prev_hash}|{_stable_json(payload)}"
return hashlib.sha256(base.encode("utf-8")).hexdigest()
def append_audit_event(db, event_type, actor, payload, request_ip=None, source="web", max_retries=5):
def get_decrypted_audit_logs(db, query=None, decrypt_fields=None):
"""
Append an audit event to a tamper-evident chain.
Retrieve and decrypt audit logs for analysis.
Args:
db: MongoDB database handle.
event_type (str): Event category.
actor (str): User/system who performed the action.
payload (dict): Event details.
request_ip (str, optional): Request origin.
source (str): Source subsystem.
query (dict): MongoDB query filter.
decrypt_fields (list): Fields within the 'payload' that should be decrypted.
"""
logs = db["audit_log"]
cursor = logs.find(query or {}).sort("chain_index", 1)
results = []
for entry in cursor:
# Decrypt specific fields if provided
if decrypt_fields:
decrypt_document_fields(entry.get("payload", {}), decrypt_fields)
results.append(entry)
return results
Returns:
dict: Inserted audit entry.
def append_audit_event(db, event_type, actor, payload, request_ip=None, source="web", max_retries=5, encrypt_fields=None):
"""
Append an audit event, optionally encrypting specific payload fields.
Args:
...
encrypt_fields (list, optional): List of keys in 'payload' to encrypt.
"""
logs = db["audit_log"]
attempts = 0
@@ -49,15 +63,24 @@ def append_audit_event(db, event_type, actor, payload, request_ip=None, source="
chain_index = int(previous.get("chain_index", 0)) + 1 if previous else 1
timestamp = datetime.datetime.utcnow()
# 1. Create the payload dictionary
event_payload = payload or {}
# 2. Encrypt sensitive fields in-place if requested
if encrypt_fields:
encrypt_document_fields(event_payload, encrypt_fields)
entry_payload = {
"event_type": event_type,
"actor": actor or "system",
"source": source,
"ip": request_ip or "",
"payload": payload or {},
"payload": event_payload,
"timestamp": timestamp.isoformat() + "Z",
}
# 3. Hash the payload (which now contains encrypted values)
entry_hash = _entry_hash(prev_hash, entry_payload)
entry = {