ge-aerospace/shopdb-flask
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Stop the audit log from being what fails the write

Browse Source 
Wave 4, second half. PUT /api/machines/<id> returned 500 and rolled the whole
edit back for any machine with a maintenance date recorded - including a rename
that never touched the date. The Machine edit form could not save such a
machine at all, which is most of them once a maintenance schedule exists.

The route builds {'old': getattr(row, field)} straight off the model, so the
changes dict carries whatever the column holds: a date, a datetime, a Decimal.
AuditLog.changes serialised it with a bare json.dumps, which raises on all
three, and nothing caught it. The audit entry and the edit share a transaction,
so the failure took the edit with it.

json.dumps now has a fallback encoder: dates and datetimes become ISO strings,
Decimals become floats, sets and bytes get something readable, and anything
else becomes its repr rather than an exception. An audit entry that reads a
little awkwardly is worth more than a failed write - the alternative, and what
happened, is losing the change AND the record of it.

Fixed at the audit log rather than in the machines route: every audited route
builds its changes dict the same way, so the next one to audit a date field
would have hit the same wall.

Four tests, including the exact shape the machines route produces. Verified by
planting the bare dumps back: two fail.

Scope: product
This commit is contained in:
cproudlock
2026-09-03 14:25:55 -04:00
parent 13e992d863
commit 5d81f50c33
3 changed files with 266 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files

351
shopdb/core/models/auditlog.py
View File

@@ -1,159 +1,192 @@
"""Audit log model for tracking changes.""" """Audit log model for tracking changes."""
from datetime import datetime, timezone from datetime import datetime, timezone
from shopdb.extensions import db from shopdb.extensions import db
def _utcnow(): def _utcnow():
# naive UTC for DB columns (stored without tzinfo) # naive UTC for DB columns (stored without tzinfo)
return datetime.now(timezone.utc).replace(tzinfo=None) return datetime.now(timezone.utc).replace(tzinfo=None)
class AuditLog(db.Model): def _jsonsafe(value):
""" """Last-resort encoder for a value the audit log was handed.
Audit log for tracking user actions.
Dates and datetimes become ISO strings, Decimals become floats, and
Records who did what, when, from where, and what changed. anything else becomes its repr rather than an exception. An audit entry
""" that reads a little awkwardly is worth more than a failed write: the
__tablename__ = 'auditlogs' alternative, and what happened, is losing the change AND the record of it.
"""
auditlogid = db.Column(db.Integer, primary_key=True) from datetime import date, datetime, time
from decimal import Decimal
# Who
userid = db.Column(db.Integer, db.ForeignKey('users.userid'), nullable=True) if isinstance(value, (datetime, date, time)):
username = db.Column(db.String(100), nullable=True) # Denormalized for history return value.isoformat()
if isinstance(value, Decimal):
# When return float(value)
timestamp = db.Column(db.DateTime, default=_utcnow, nullable=False, index=True) if isinstance(value, (set, frozenset)):
return sorted(str(item) for item in value)
# Where (client info) if isinstance(value, bytes):
ipaddress = db.Column(db.String(45), nullable=True) # IPv6 max length return value.decode('utf-8', 'replace')
useragent = db.Column(db.String(255), nullable=True) return repr(value)
# What
action = db.Column(db.String(20), nullable=False, index=True) # created, updated, deleted class AuditLog(db.Model):
entitytype = db.Column(db.String(50), nullable=False, index=True) # Asset, Printer, Setting, etc. """
entityid = db.Column(db.Integer, nullable=True) # ID of the affected record Audit log for tracking user actions.
entityname = db.Column(db.String(255), nullable=True) # Human-readable identifier
Records who did what, when, from where, and what changed.
# Changes (stored as JSON text) """
_changes = db.Column('changes', db.Text, nullable=True) # {"field": {"old": x, "new": y}, ...} __tablename__ = 'auditlogs'
@property auditlogid = db.Column(db.Integer, primary_key=True)
def changes(self):
"""Get changes as dict.""" # Who
if self._changes: userid = db.Column(db.Integer, db.ForeignKey('users.userid'), nullable=True)
import json username = db.Column(db.String(100), nullable=True) # Denormalized for history
try:
return json.loads(self._changes) # When
except (json.JSONDecodeError, TypeError): timestamp = db.Column(db.DateTime, default=_utcnow, nullable=False, index=True)
return None
return None # Where (client info)
ipaddress = db.Column(db.String(45), nullable=True) # IPv6 max length
@changes.setter useragent = db.Column(db.String(255), nullable=True)
def changes(self, value):
"""Set changes from dict.""" # What
if value is not None: action = db.Column(db.String(20), nullable=False, index=True) # created, updated, deleted
import json entitytype = db.Column(db.String(50), nullable=False, index=True) # Asset, Printer, Setting, etc.
self._changes = json.dumps(value) entityid = db.Column(db.Integer, nullable=True) # ID of the affected record
else: entityname = db.Column(db.String(255), nullable=True) # Human-readable identifier
self._changes = None
# Changes (stored as JSON text)
# Additional context _changes = db.Column('changes', db.Text, nullable=True) # {"field": {"old": x, "new": y}, ...}
details = db.Column(db.Text, nullable=True) # Optional description
@property
# Relationship def changes(self):
user = db.relationship('User', backref=db.backref('auditlogs', lazy='dynamic')) """Get changes as dict."""
if self._changes:
def to_dict(self): import json
return { try:
'auditlogid': self.auditlogid, return json.loads(self._changes)
'userid': self.userid, except (json.JSONDecodeError, TypeError):
'username': self.username, return None
'timestamp': self.timestamp.isoformat() + 'Z' if self.timestamp else None, return None
'ipaddress': self.ipaddress,
'useragent': self.useragent, @changes.setter
'action': self.action, def changes(self, value):
'entitytype': self.entitytype, """Set changes from dict.
'entityid': self.entityid,
'entityname': self.entityname, Serialised with a fallback for the types a model actually holds. A bare
'changes': self.changes, json.dumps raises on a date, a datetime or a Decimal, and the caller is
'details': self.details a route building {'old': getattr(row, field)} straight off the model -
} so the moment an audited field held a date, the whole request died.
@classmethod That is not hypothetical: PUT /api/machines/<id> returned 500 and rolled
def log(cls, action: str, entitytype: str, entityid: int = None, the edit back for ANY machine with a maintenance date recorded,
entityname: str = None, changes: dict = None, details: str = None, including a rename that never touched the date. The audit log refusing
user=None, request=None): to record a change must not be what stops the change.
""" """
Create an audit log entry. if value is not None:
import json
Args: self._changes = json.dumps(value, default=_jsonsafe)
action: 'created', 'updated', 'deleted' else:
entitytype: Type of entity (e.g., 'Asset', 'Printer', 'Setting') self._changes = None
entityid: ID of the affected record
entityname: Human-readable name/identifier # Additional context
changes: Dict of field changes {"field": {"old": x, "new": y}} details = db.Column(db.Text, nullable=True) # Optional description
details: Optional description
user: Current user object (or will try to get from flask-jwt-extended) # Relationship
request: Flask request object (or will try to get current request) user = db.relationship('User', backref=db.backref('auditlogs', lazy='dynamic'))
"""
from flask import request as flask_request def to_dict(self):
from flask_jwt_extended import current_user, verify_jwt_in_request return {
'auditlogid': self.auditlogid,
# Get user info 'userid': self.userid,
if user is None: 'username': self.username,
try: 'timestamp': self.timestamp.isoformat() + 'Z' if self.timestamp else None,
verify_jwt_in_request(optional=True) 'ipaddress': self.ipaddress,
user = current_user 'useragent': self.useragent,
except: 'action': self.action,
pass 'entitytype': self.entitytype,
'entityid': self.entityid,
userid = user.userid if user else None 'entityname': self.entityname,
username = user.username if user else None 'changes': self.changes,
'details': self.details
# Get request info }
request_obj = request or flask_request
ipaddress = None @classmethod
useragent = None def log(cls, action: str, entitytype: str, entityid: int = None,
entityname: str = None, changes: dict = None, details: str = None,
if request_obj: user=None, request=None):
# First-hop client IP, port stripped (ARR forwards clientip:port). """
from shopdb.utils.clientip import client_ip Create an audit log entry.
ipaddress = client_ip(request_obj)
useragent = request_obj.headers.get('User-Agent', '')[:255] Args:
action: 'created', 'updated', 'deleted'
entry = cls( entitytype: Type of entity (e.g., 'Asset', 'Printer', 'Setting')
userid=userid, entityid: ID of the affected record
username=username, entityname: Human-readable name/identifier
ipaddress=ipaddress, changes: Dict of field changes {"field": {"old": x, "new": y}}
useragent=useragent, details: Optional description
action=action, user: Current user object (or will try to get from flask-jwt-extended)
entitytype=entitytype, request: Flask request object (or will try to get current request)
entityid=entityid, """
entityname=entityname, from flask import request as flask_request
changes=changes, from flask_jwt_extended import current_user, verify_jwt_in_request
details=details
) # Get user info
if user is None:
db.session.add(entry) try:
# Don't commit here - let the caller handle transaction verify_jwt_in_request(optional=True)
return entry user = current_user
except:
@classmethod pass
def log_create(cls, entitytype: str, entity, name_field: str = 'name'):
"""Log a create action.""" userid = user.userid if user else None
entityname = getattr(entity, name_field, None) or str(entity) username = user.username if user else None
entityid = getattr(entity, f'{entitytype.lower()}id', None) or getattr(entity, 'id', None)
return cls.log('created', entitytype, entityid=entityid, entityname=entityname) # Get request info
request_obj = request or flask_request
@classmethod ipaddress = None
def log_update(cls, entitytype: str, entity, changes: dict, name_field: str = 'name'): useragent = None
"""Log an update action with changes."""
entityname = getattr(entity, name_field, None) or str(entity) if request_obj:
entityid = getattr(entity, f'{entitytype.lower()}id', None) or getattr(entity, 'id', None) # First-hop client IP, port stripped (ARR forwards clientip:port).
return cls.log('updated', entitytype, entityid=entityid, entityname=entityname, changes=changes) from shopdb.utils.clientip import client_ip
ipaddress = client_ip(request_obj)
@classmethod useragent = request_obj.headers.get('User-Agent', '')[:255]
def log_delete(cls, entitytype: str, entityid: int, entityname: str = None):
"""Log a delete action.""" entry = cls(
return cls.log('deleted', entitytype, entityid=entityid, entityname=entityname) userid=userid,
username=username,
ipaddress=ipaddress,
useragent=useragent,
action=action,
entitytype=entitytype,
entityid=entityid,
entityname=entityname,
changes=changes,
details=details
)
db.session.add(entry)
# Don't commit here - let the caller handle transaction
return entry
@classmethod
def log_create(cls, entitytype: str, entity, name_field: str = 'name'):
"""Log a create action."""
entityname = getattr(entity, name_field, None) or str(entity)
entityid = getattr(entity, f'{entitytype.lower()}id', None) or getattr(entity, 'id', None)
return cls.log('created', entitytype, entityid=entityid, entityname=entityname)
@classmethod
def log_update(cls, entitytype: str, entity, changes: dict, name_field: str = 'name'):
"""Log an update action with changes."""
entityname = getattr(entity, name_field, None) or str(entity)
entityid = getattr(entity, f'{entitytype.lower()}id', None) or getattr(entity, 'id', None)
return cls.log('updated', entitytype, entityid=entityid, entityname=entityname, changes=changes)
@classmethod
def log_delete(cls, entitytype: str, entityid: int, entityname: str = None):
"""Log a delete action."""
return cls.log('deleted', entitytype, entityid=entityid, entityname=entityname)

41
tests/test_core/test_strict_write_errors.py
View File

@@ -97,3 +97,44 @@ def test_the_strict_sql_mode_is_actually_set_on_the_mysql_connection():
connect_args = Config.SQLALCHEMY_ENGINE_OPTIONS['connect_args'] connect_args = Config.SQLALCHEMY_ENGINE_OPTIONS['connect_args']
assert 'STRICT_TRANS_TABLES' in connect_args['init_command'] assert 'STRICT_TRANS_TABLES' in connect_args['init_command']
# --- the audit log must never be what fails the write ------------------------
def test_audit_changes_serialise_the_types_a_model_actually_holds():
"""A route builds {'old': getattr(row, field)} straight off the model, so
the changes dict carries dates, datetimes and Decimals. json.dumps raises
on all three, and the exception took the whole request down: PUT
/api/machines/<id> 500d and rolled back for any machine with a maintenance
date, including a rename that never touched it."""
import datetime
import decimal
from shopdb.core.models.auditlog import AuditLog
entry = AuditLog()
entry.changes = {
'nextmaintenancedate': {'old': datetime.date(2026, 12, 15),
'new': '2027-01-05'},
'lastseen': {'old': datetime.datetime(2026, 1, 2, 3, 4, 5),
'new': None},
'cost': {'old': decimal.Decimal('12.50'), 'new': 13},
}
stored = entry.changes
assert stored['nextmaintenancedate']['old'] == '2026-12-15'
assert stored['lastseen']['old'] == '2026-01-02T03:04:05'
assert stored['cost']['old'] == 12.5
def test_an_unserialisable_value_is_recorded_rather_than_raising():
"""An audit entry that reads awkwardly beats a failed write: the
alternative is losing the change AND the record of it."""
from shopdb.core.models.auditlog import AuditLog
class Odd:
def __repr__(self):
return '<Odd value>'
entry = AuditLog()
entry.changes = {'thing': {'old': Odd(), 'new': None}}
assert entry.changes['thing']['old'] == '<Odd value>'

33
tests/test_plugins/test_measuringtools.py
View File

@@ -553,3 +553,36 @@ def test_a_mistyped_date_is_refused_at_create_too(client, auth_headers):
listed = client.get('/api/measuringtools?search=MT-DATE-3', listed = client.get('/api/measuringtools?search=MT-DATE-3',
headers=auth_headers).get_json()['data'] headers=auth_headers).get_json()['data']
assert listed == [], 'nothing may be created by a refused request' assert listed == [], 'nothing may be created by a refused request'
# --- an audited field holding a date must not kill the request ---------------
#
# PUT /api/machines/<id> returned 500 and rolled the whole edit back for any
# machine with a maintenance date: the route builds
# {'old': getattr(row, field)} straight off the model, and json.dumps raises on
# a date. A rename that never touched the date failed too. The audit log
# refusing to record a change must not be what stops the change.
def test_an_edit_is_saved_even_when_an_audited_field_holds_a_date(
client, auth_headers):
created = client.post('/api/measuringtools', headers=auth_headers, json={
'assetnumber': 'MT-AUDIT-1',
'name': 'Before rename',
'statusid': _status_id(client),
'lastcalibrationdate': '2026-01-15',
'nextcalibrationdate': '2027-01-15',
})
assert created.status_code == 201, created.get_json()
toolid = created.get_json()['data']['measuringtool']['measuringtoolid']
# A rename that does not touch the dates. The date is only in the CHANGES
# dict because the interval moves with it.
renamed = client.put(f'/api/measuringtools/{toolid}', headers=auth_headers,
json={'name': 'After rename',
'nextcalibrationdate': '2028-01-15'})
assert renamed.status_code == 200, renamed.get_json()
after = client.get(f'/api/measuringtools/{toolid}',
headers=auth_headers).get_json()['data']
assert after['name'] == 'After rename'
assert after['measuringtool']['nextcalibrationdate'] == '2028-01-15'