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

Remediate the 2026-09-03 feature-test findings

Browse Source
This commit is contained in:
Cameron Proudlock
2026-09-03 17:15:03 -04:00
parent cf95daf3bb
commit fd513c08eb
112 changed files with 5182 additions and 857 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
shopdb/core/api/apitokens.py
View File

@@ -163,7 +163,18 @@ def update_apitoken(tokenid: int):
return error_response(ErrorCodes.VALIDATION_ERROR, 'name cannot be empty')
token.name = newname
if 'isactive' in data:
token.isactive = bool(data['isactive'])
# Revocation is ONE WAY. A revoked token's secret is out there - that
# is why it was revoked - and re-activating the row made the same
# secret work again, so "revoked" meant "paused until somebody flips it
# back". Deactivating through this route is still allowed; bringing one
# back is not, and the answer says to mint a new one.
wanted = bool(data['isactive'])
if wanted and not token.isactive:
return error_response(
ErrorCodes.VALIDATION_ERROR,
'A revoked token cannot be re-activated - its secret is '
'already out. Create a new token instead.')
token.isactive = wanted
if 'scopes' in data:
scopelist, scope_error = _validate_scopes(data.get('scopes'), token.user)
if scope_error is not None:

39
shopdb/core/api/applications.py
View File

@@ -2,6 +2,7 @@
import glob
import hashlib
import re
import os
from flask import Blueprint, request, current_app
@@ -20,6 +21,7 @@ from shopdb.utils.responses import (
)
from shopdb.utils.pagination import get_pagination_params, paginate_query
from shopdb.utils.import_mode import apply_import_timestamps
from shopdb.utils.datevalues import DateValueError, parse_date
# Uploads live in the instance dir, like model images and part photos, so a
@@ -326,6 +328,24 @@ def create_version(app_id: int):
if not data or not data.get('version'):
return error_response(ErrorCodes.VALIDATION_ERROR, 'version is required')
# A digest is a claim the delivery path CHECKS against, so a wrong one is
# worse than none: an over-long value was silently truncated to the column
# width and came back as 64 plausible hex characters, indistinguishable
# from a real SHA-256. Either it is a SHA-256 or it is refused.
digest = (data.get('sha256') or '').strip().lower()
if digest and not re.fullmatch(r'[0-9a-f]{64}', digest):
return error_response(
ErrorCodes.VALIDATION_ERROR,
'sha256 must be 64 hexadecimal characters')
# Refused rather than stored raw: a malformed date used to reach MySQL,
# which (before STRICT_TRANS_TABLES) wrote 0000-00-00 and then broke every
# read of this application.
try:
releasedate = parse_date(data.get('releasedate'), 'releasedate')
except DateValueError as bad:
return error_response(ErrorCodes.VALIDATION_ERROR, str(bad))
if AppVersion.query.filter_by(appid=app_id, version=data['version']).first():
return error_response(
ErrorCodes.CONFLICT,
@@ -336,9 +356,9 @@ def create_version(app_id: int):
version = AppVersion(
appid=app_id,
version=data['version'],
releasedate=data.get('releasedate'),
releasedate=releasedate,
notes=data.get('notes'),
sha256=data.get('sha256'),
sha256=digest or None,
silentargs=data.get('silentargs')
)
@@ -558,9 +578,20 @@ def upload_application_image(app_id: int):
@applications_bp.route('/image/<path:filename>', methods=['GET'])
@require_permission('applications.view')
def serve_application_image(filename):
"""Serve an uploaded application image (public - lists and tiles read it)."""
"""Serve an uploaded application image.
Ungated, like every other image route in the product (the branding logo,
the map blueprint, an employee photo, a printed-part picture). A browser
fetching <img src="/api/applications/image/application-3.png"> sends no
Authorization header, so a permission check here does not protect the file
- it just guarantees the picture never appears. The tile fell back to its
broken-image handler on every page, including for an admin.
What is exposed is an application icon to somebody who already knows the
filename, which is the same posture as its four siblings. Registered in
tests/test_core/test_public_read_surface.py with that reasoning.
"""
return send_upload(_appimage_dir(), filename)

68
shopdb/core/api/assets.py
View File

@@ -13,6 +13,7 @@ from shopdb.utils.responses import (
ErrorCodes
)
from shopdb.utils.pagination import get_pagination_params, paginate_query
from shopdb.utils.queryargs import intarg
from shopdb.utils.authz import require_permission
from shopdb.utils.import_mode import apply_import_timestamps
@@ -365,20 +366,20 @@ def list_assets():
query = query.join(AssetType).filter(AssetType.assettype == type_name)
# Type filter by ID
if type_id := request.args.get('typeid', request.args.get('type_id')):
query = query.filter(Asset.assettypeid == int(type_id))
if (type_id := intarg('typeid', intarg('type_id'))) is not None:
query = query.filter(Asset.assettypeid == type_id)
# Status filter
if status_id := request.args.get('statusid', request.args.get('status_id')):
query = query.filter(Asset.statusid == int(status_id))
if (status_id := intarg('statusid', intarg('status_id'))) is not None:
query = query.filter(Asset.statusid == status_id)
# Location filter
if location_id := request.args.get('locationid', request.args.get('location_id')):
query = query.filter(Asset.locationid == int(location_id))
if (location_id := intarg('locationid', intarg('location_id'))) is not None:
query = query.filter(Asset.locationid == location_id)
# Business unit filter
if bu_id := request.args.get('businessunitid', request.args.get('businessunit_id')):
query = query.filter(Asset.businessunitid == int(bu_id))
if (bu_id := intarg('businessunitid', intarg('businessunit_id'))) is not None:
query = query.filter(Asset.businessunitid == bu_id)
# Sorting
sort_by = request.args.get('sort', 'assetnumber')
@@ -1011,17 +1012,26 @@ def get_assets_map():
Asset.mapy.isnot(None)
)
# Dualpath single-machine collapse: hide the secondary bay marker so a
# dual-bay pair shows one marker; the primary carries the partner label.
# Gated on the site setting (default on).
# Dualpath single-machine collapse on the map: a secondary bay draws its
# OWN marker when somebody has positioned it, and is represented by the
# primary's marker when nobody has.
#
# It used to be excluded unconditionally, which made setting a position on
# a secondary bay a silent no-op: the form accepted the pin, stored it, and
# the map never drew it. One marker per pair is right when the pair sits in
# one place, which is the common case and still the default - but a
# dual-bay machine whose second bay is physically somewhere else is exactly
# what positioning it separately means.
#
# No extra filtering is needed for the fallback half: the query above
# already requires mapx/mapy, so an unpositioned secondary never reaches
# here and the primary's marker stands for both, carrying the combined
# label.
from shopdb.core.services.dualpath import (
resolve_dualpath_pairs, dualpath_single_machine_enabled)
dualpath_collapse = None
if dualpath_single_machine_enabled():
dualpath_collapse = resolve_dualpath_pairs()
if dualpath_collapse.secondaryassetids:
query = query.filter(
Asset.assetid.notin_(dualpath_collapse.secondaryassetids))
selected_assettype = request.args.get('assettype')
@@ -1030,8 +1040,7 @@ def get_assets_map():
query = query.join(AssetType).filter(AssetType.assettype == selected_assettype)
# Filter by subtype (depends on asset type) - case-insensitive matching
if subtype_id := request.args.get('subtype'):
subtype_id = int(subtype_id)
if (subtype_id := intarg('subtype')) is not None:
# Normalize the underscore DB form (measuring_tool, network_device) to
# the space form the branches below compare against.
asset_type_lower = (
@@ -1079,16 +1088,16 @@ def get_assets_map():
pass
# Filter by business unit
if bu_id := request.args.get('businessunitid'):
query = query.filter(Asset.businessunitid == int(bu_id))
if (bu_id := intarg('businessunitid')) is not None:
query = query.filter(Asset.businessunitid == bu_id)
# Filter by status
if status_id := request.args.get('statusid'):
query = query.filter(Asset.statusid == int(status_id))
if (status_id := intarg('statusid')) is not None:
query = query.filter(Asset.statusid == status_id)
# Filter by location
if location_id := request.args.get('locationid'):
query = query.filter(Asset.locationid == int(location_id))
if (location_id := intarg('locationid')) is not None:
query = query.filter(Asset.locationid == location_id)
# Search filter
if search := request.args.get('search'):
@@ -1102,6 +1111,10 @@ def get_assets_map():
assets = query.all()
# Every asset that ends up with a marker. Used to decide whether a dual-bay
# partner draws its own pin or is represented by this one.
positioned_assetids = {asset.assetid for asset in assets}
# Batch-load primary IPs in a single query instead of N+1 per asset.
# Prefer isprimary=True IP, fall back to any IP (comtypeid=1).
asset_ids = [a.assetid for a in assets]
@@ -1153,10 +1166,17 @@ def get_assets_map():
if type_data:
item['typedata'] = type_data
# combined-label partner for a collapsed dual-bay pair (label only)
# Combined label for a dual-bay pair, and ONLY when this marker really
# stands for both bays. A partner that draws its own marker gets its
# own name: labelling two separate pins '2007 / 2008' would say the
# same machine is in two places.
if dualpath_collapse:
partner = dualpath_collapse.partnerbyasset.get(asset.assetid)
item['dualpathpartner'] = partner['assetnumber'] if partner else None
partner_has_marker = bool(
partner and partner['assetid'] in positioned_assetids)
item['dualpathpartner'] = (
partner['assetnumber'] if partner and not partner_has_marker
else None)
data.append(item)

18
shopdb/core/api/auditlogs.py
View File

@@ -5,7 +5,8 @@ from flask_jwt_extended import jwt_required
from shopdb.core.models import AuditLog
from shopdb.extensions import db
from shopdb.utils.responses import success_response, error_response, ErrorCodes
from shopdb.utils.responses import (success_response, error_response,
paginated_response, ErrorCodes)
from shopdb.utils.authz import require_permission
auditlogs_bp = Blueprint('auditlogs', __name__)
@@ -123,15 +124,12 @@ def list_auditlogs():
for r in rows:
r['userfullname'] = fullnames.get(str(r.get('username')))
return success_response(
rows,
meta={
'page': page,
'perpage': perpage,
'total': pagination.total,
'pages': pagination.pages
}
)
# paginated_response, like every other list in the product. This one built
# its own meta with the fields FLAT and named 'pages', so a client written
# against the documented meta.pagination.{page,perpage,total,totalpages}
# found nothing where it looked - on the one list whose whole purpose is to
# be read by somebody checking what happened.
return paginated_response(rows, page, perpage, pagination.total)
@auditlogs_bp.route('/entity/<entitytype>/<int:entityid>', methods=['GET'])

35
shopdb/core/api/maplevels.py
View File

@@ -36,10 +36,27 @@ def _blueprint_dir():
# =============================================================================
# Read - public
#
# ADR-017 rule 6: the level list is readable without authentication, because the
# printer installer map runs optional-auth before anyone logs in and needs a
# blueprint. It follows the precedent of printer install-list and the slide
# feed, both still public.
#
# afce2437 ("Reading anything needs a permission") swept 165 open GET routes
# and caught these three in the net. Fail-closed, so nothing leaked - but the
# anonymous installer map has drawn no floor plan since, while this same file
# kept the docstring saying the reads are public, the API inventory kept
# declaring auth "none", and PrinterInstallerMap.vue kept its comment
# explaining why. Four places agreed and the code did not.
#
# Restored rather than amending the ADR: blueprint paths and level names are
# not secrets, the asset positions on them already render on public kiosk
# pages, and the feature the rule exists for is broken without it. These three
# are registered in tests/test_core/test_public_read_surface.py.
# =============================================================================
@maplevels_bp.route('', methods=['GET'])
@require_permission('assets.view')
@jwt_required(optional=True)
def list_levels():
"""Every active level, grouped by building, in display order.
@@ -74,7 +91,7 @@ def list_levels():
@maplevels_bp.route('/<int:levelid>', methods=['GET'])
@require_permission('assets.view')
@jwt_required(optional=True)
def get_level(levelid):
level = db.session.get(MapLevel, levelid)
if not level or not level.isactive:
@@ -84,7 +101,7 @@ def get_level(levelid):
@maplevels_bp.route('/<int:levelid>/blueprint/<path:filename>', methods=['GET'])
@require_permission('assets.view')
@jwt_required(optional=True)
def serve_blueprint(filename, levelid=None):
"""Serve a level's blueprint image. Public, for the same reason as above."""
return send_upload(_blueprint_dir(), filename)
@@ -160,9 +177,17 @@ def create_level():
data = request.get_json() or {}
name = (data.get('levelname') or '').strip()
buildingid = data.get('buildingid')
# Names the fields this route actually reads. It said 'buildingname',
# which this handler never looks at, so somebody who omitted the level name
# was told to supply something else entirely.
if not name or not buildingid:
return error_response(ErrorCodes.VALIDATION_ERROR,
'buildingname and buildingid are required')
missing = [field for field, value in
(('levelname', name), ('buildingid', buildingid))
if not value]
return error_response(
ErrorCodes.VALIDATION_ERROR,
'%s %s required' % (' and '.join(missing),
'is' if len(missing) == 1 else 'are'))
if not db.session.get(Building, buildingid):
return error_response(ErrorCodes.NOT_FOUND, 'No such building',
http_code=404)

1
shopdb/core/api/models.py
View File

@@ -235,7 +235,6 @@ def upload_model_image(model_id: int):
@models_bp.route('/image/<path:filename>', methods=['GET'])
@require_permission('assets.view')
def serve_model_image(filename):
"""Serve an uploaded model image (public - asset detail pages read it)."""
return send_upload(_modelimage_dir(), filename)

35
shopdb/core/api/reports.py
View File

@@ -13,6 +13,7 @@ from shopdb.core.models import (
)
from shopdb.utils.responses import success_response, error_response, ErrorCodes
from shopdb.utils.authz import require_permission
from shopdb.utils.queryargs import intarg
reports_bp = Blueprint('reports', __name__)
@@ -62,8 +63,8 @@ def machines_by_type():
)
# Business unit filter
if bu_id := request.args.get('businessunitid'):
query = query.filter(Asset.businessunitid == int(bu_id))
if (bu_id := intarg('businessunitid')) is not None:
query = query.filter(Asset.businessunitid == bu_id)
# Dualpath single-machine collapse: exclude the secondary bay so a dual-bay
# pair counts once per type bucket. Gated on the site setting (default on).
@@ -128,12 +129,12 @@ def assets_by_status():
)
# Asset type filter
if type_id := request.args.get('assettypeid'):
query = query.filter(Asset.assettypeid == int(type_id))
if (type_id := intarg('assettypeid')) is not None:
query = query.filter(Asset.assettypeid == type_id)
# Business unit filter
if bu_id := request.args.get('businessunitid'):
query = query.filter(Asset.businessunitid == int(bu_id))
if (bu_id := intarg('businessunitid')) is not None:
query = query.filter(Asset.businessunitid == bu_id)
query = query.group_by(
AssetStatus.statusid,
@@ -175,7 +176,7 @@ def kb_popularity():
- limit: Number of results (default 20)
- format: 'json' (default) or 'csv'
"""
limit = min(int(request.args.get('limit', 20)), 100)
limit = min(intarg('limit', 20), 100)
# Knowledge Base is a plugin; degrade gracefully if it is not installed.
try:
@@ -362,10 +363,10 @@ def asset_inventory():
AssetType.isactive == True
)
if bu_id := request.args.get('businessunitid'):
by_type = by_type.filter(Asset.businessunitid == int(bu_id))
if loc_id := request.args.get('locationid'):
by_type = by_type.filter(Asset.locationid == int(loc_id))
if (bu_id := intarg('businessunitid')) is not None:
by_type = by_type.filter(Asset.businessunitid == bu_id)
if (loc_id := intarg('locationid')) is not None:
by_type = by_type.filter(Asset.locationid == loc_id)
by_type = by_type.group_by(AssetType.assettype).all()
@@ -380,10 +381,10 @@ def asset_inventory():
AssetStatus.isactive == True
)
if bu_id := request.args.get('businessunitid'):
by_status = by_status.filter(Asset.businessunitid == int(bu_id))
if loc_id := request.args.get('locationid'):
by_status = by_status.filter(Asset.locationid == int(loc_id))
if (bu_id := intarg('businessunitid')) is not None:
by_status = by_status.filter(Asset.businessunitid == bu_id)
if (loc_id := intarg('locationid')) is not None:
by_status = by_status.filter(Asset.locationid == loc_id)
by_status = by_status.group_by(AssetStatus.status, AssetStatus.color).all()
@@ -398,8 +399,8 @@ def asset_inventory():
Location.isactive == True
)
if bu_id := request.args.get('businessunitid'):
by_location = by_location.filter(Asset.businessunitid == int(bu_id))
if (bu_id := intarg('businessunitid')) is not None:
by_location = by_location.filter(Asset.businessunitid == bu_id)
by_location = by_location.group_by(Location.locationname).all()

25
shopdb/core/api/search.py
View File

@@ -244,6 +244,18 @@ def _get_asset_result(asset, query, relevance=None):
if relevance is None:
relevance = 15
# A retired asset never outranks a live one, whatever it matched on: the
# person searching almost always wants the machine that is still on the
# floor. Flagged too, so the result can say so rather than looking like any
# other hit.
# getattr, because this builder is also called with lightweight
# stand-ins that carry only the fields a result needs. An object that does
# not say is treated as live, which is the safe direction: a live asset
# ranked correctly beats a live asset buried as retired.
retired = getattr(asset, 'isactive', True) is False
if retired:
relevance = max(1, relevance - 60)
return {
'type': asset_type_name,
'id': result_id,
@@ -251,7 +263,8 @@ def _get_asset_result(asset, query, relevance=None):
'subtitle': subtitle,
'location': location_name,
'url': url,
'relevance': relevance
'relevance': relevance,
'retired': retired
}
@@ -499,11 +512,17 @@ def _search_assets(query, search_term):
joinedload(Asset.assettype),
joinedload(Asset.location),
).filter(
Asset.isactive == True,
# Retired assets are INCLUDED, and ranked below live ones. Retire
# is a soft delete whose confirm dialog promises the record can be
# restored - and the one place somebody looks for a machine they
# retired last month is the search box. Excluding them meant the
# record existed, was reachable by URL, and could not be found by
# anybody who did not already know where it was.
_word_match(query, Asset.assetnumber, Asset.name,
Asset.serialnumber, Asset.notes,
Asset.gaugelabreference, Asset.maintenancereference)
).order_by(Asset.assetnumber, Asset.assetid).limit(15).all()
).order_by(Asset.isactive.desc(), Asset.assetnumber,
Asset.assetid).limit(15).all()
for asset in assets:
relevance = 15

43
shopdb/core/api/settings.py
View File

@@ -811,6 +811,34 @@ def update_setting(key: str):
if _is_secret(key) and value == SECRET_MASK:
return success_response(_serialize_setting(setting), message='Setting unchanged')
# The declared valuetype is a promise about what every reader will get, and
# nothing checked it: "A6-not-a-number" was stored verbatim on an integer
# key and every consumer then read 0, because Setting.get_typed_value
# swallows the ValueError. The setting looked set, the behaviour was the
# default, and nothing said so.
valuetype = (setting.valuetype or 'string').lower()
if value is not None and not isinstance(value, bool):
text = str(value).strip()
if valuetype == 'integer' and text:
try:
int(text)
except ValueError:
return error_response(
ErrorCodes.VALIDATION_ERROR,
'%s is a whole-number setting (got %r)' % (key, value))
elif valuetype == 'float' and text:
try:
float(text)
except ValueError:
return error_response(
ErrorCodes.VALIDATION_ERROR,
'%s is a number setting (got %r)' % (key, value))
elif valuetype == 'boolean' and text:
if text.lower() not in ('true', 'false', '1', '0', 'yes', 'no'):
return error_response(
ErrorCodes.VALIDATION_ERROR,
'%s is a true/false setting (got %r)' % (key, value))
# Convert value to string for storage
if isinstance(value, bool):
setting.value = 'true' if value else 'false'
@@ -1460,14 +1488,13 @@ def build_default_settings():
'Incoming Webhook), adaptivecard (Teams Workflow), '
'or json (generic {title,text})'
},
# Audit log settings
{
'key': 'audit_retention_days',
'value': '90',
'valuetype': 'integer',
'category': 'audit',
'description': 'Number of days to retain audit logs (0 = keep forever)'
},
# Audit log retention lives at 'audit_retentiondays' (category site),
# defined above and read by shopdb/core/services/auditretention.py. A
# second key 'audit_retention_days' was seeded here with a different
# default and a different category, and the Audit & Logging page wrote
# to THAT one - so the page saved a value, read it back, and nothing
# ever consulted it. Removed rather than aliased: two keys for one
# setting is how this happened.
# Authentication settings
{
'key': 'saml_enabled',

21
shopdb/core/api/users.py
View File

@@ -187,9 +187,28 @@ def update_user(userid: int):
user.failedlogins = 0
changes['unlocked'] = {'old': True, 'new': False}
# Password change
# Password change.
#
# An ADMIN may set one here - that is how a temporary password is issued,
# and mustchangepassword forces the holder to replace it at next login.
# Somebody changing their OWN password may not: this route asks for no
# current password and enforces no length, so it was a way around
# /api/auth/change-password, which requires both. The two rules have to
# agree or the weaker one is the real one.
if 'password' in data and data['password']:
if not current_user.hasrole('admin'):
return error_response(
ErrorCodes.FORBIDDEN,
'Change your own password at /api/auth/change-password, which '
'requires your current one.',
http_code=403)
if len(data['password']) < 8:
return error_response(
ErrorCodes.VALIDATION_ERROR,
'Password must be at least 8 characters')
user.passwordhash = generate_password_hash(data['password'])
# Issued by somebody else, so it is temporary by definition.
user.mustchangepassword = True
changes['password'] = {'old': '***', 'new': '***'}
if changes:

18
shopdb/core/models/asset.py
View File

@@ -256,6 +256,24 @@ class Asset(BaseModel, SoftDeleteMixin, AuditMixin):
elif hasattr(self, 'measuringtool') and self.measuringtool:
result['pluginid'] = self.measuringtool.measuringtoolid
# What this asset OWNS, before any inheritance is merged in below.
#
# The merge is right for a reader: a detail page, a list and the map all
# want the resolved location, whoever it came from. It is wrong for a
# FORM, which reads the same payload and cannot tell an owned value
# from a derived one - so the PC form pre-selected an inherited
# location, the operator saved something else entirely, and the
# inherited value was written back as the record's own. The asset then
# stopped following its parent, silently, on an edit that never touched
# location.
#
# Always emitted, even when nothing is inherited, so a client can bind
# to these without checking whether the merge happened.
result['ownlocationid'] = self.locationid
result['ownmapx'] = self.mapx
result['ownmapy'] = self.mapy
result['ownlevelid'] = self.levelid
# Include inherited location if this asset has no location data
if include_inherited_location:
inherited = self.get_inherited_location()

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

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

9
shopdb/core/models/base.py
View File

@@ -1,6 +1,6 @@
"""Base model class with common fields."""
from datetime import datetime, timezone
from datetime import date, datetime, timezone
from shopdb.extensions import db
@@ -63,6 +63,13 @@ class BaseModel(db.Model):
value = getattr(self, c.name)
if isinstance(value, datetime):
value = value.isoformat() + 'Z'
elif isinstance(value, date):
# A date, not an instant: no time and no timezone marker. Left
# to Flask's encoder it became an HTTP-date
# ("Thu, 15 Jan 2026 00:00:00 GMT"), which a browser parses
# back into a moment and can render as the day before.
# datetime is a subclass of date, so this must stay second.
value = value.isoformat()
result[c.name] = value
return result