afce24379c
A sweep of the URL map found 165 GET routes that answered anybody, with no token at all: the whole asset inventory, every machine, PC, printer and network device, the USB checkout history and who is holding which drive, the staff directory with its photographs, every core report. Guest access had been argued about one feature-flag at a time while the front door stood open. Each of those now carries the read permission it belongs to. That also makes the guest role mean something: a site can open exactly the reads it chooses, and nothing follows by accident. Thirty-three endpoints stay public, and each is named with its reason rather than left to be inferred: the app shell and its bootstrap, the settings allowlist (the handler already filters an anonymous caller down to branding and a few keys), images fetched by <img src> which carries no Authorization header, what a display kiosk renders before anybody signs in, the parts and USB kiosks, the client scripts that run as SYSTEM before a user session exists, and a secure-send link whose unguessable token IS the authorisation. tests/test_core/test_public_read_surface.py is the standing gate, the companion to the mutating-route sweep that has guarded writes for months. A new public read fails CI until somebody adds it to that list in front of a reviewer. Four further reads need a login but no permission, because the handler decides from who you are - your own user record, your own API tokens, the role catalogue the grid renders - and the same test proves each still refuses an anonymous caller. Verified not vacuous by ungating one route and watching the sweep name it. The employees plugin gained employees.view; the directory is names, SSOs, photographs and reporting lines, and it had no permission of any kind. Around ninety tests asserted payload shapes against endpoints that answered anybody. They are about what the answer says, not about who may ask, so they are signed in now rather than the gate being loosened. The ones that assert an anonymous caller is refused were left exactly as they were.
355 lines
15 KiB
Python
355 lines
15 KiB
Python
"""Buildings and levels: the drawings a marker can be placed on (ADR-017).
|
|
|
|
Reads are UNAUTHENTICATED. The printer installer map runs before anyone logs in
|
|
and needs a blueprint to draw, exactly as the printer install-list and the slide
|
|
feed already do. A level name, a blueprint path and a pixel size are not
|
|
secrets - the marker positions drawn on them already render on kiosk pages.
|
|
|
|
Writes require admin: adding a level changes where every marker on it appears.
|
|
"""
|
|
|
|
import os
|
|
|
|
from flask import Blueprint, request, current_app
|
|
from flask_jwt_extended import jwt_required
|
|
from werkzeug.utils import secure_filename
|
|
|
|
from shopdb.extensions import db
|
|
from shopdb.core.models import Building, MapLevel, Asset, AuditLog
|
|
from shopdb.utils.responses import success_response, error_response, ErrorCodes
|
|
from shopdb.utils.authz import require_role
|
|
from shopdb.api import send_upload
|
|
from shopdb.utils.imagesize import image_size
|
|
from shopdb.utils.authz import require_permission
|
|
|
|
maplevels_bp = Blueprint('maplevels', __name__)
|
|
|
|
# Same set the map blueprint upload already accepts. SVG stays allowed because a
|
|
# floor plan is vector by nature; it is served through send_upload, which sends
|
|
# the sandbox headers that stop one executing as script.
|
|
BLUEPRINT_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.svg'}
|
|
|
|
|
|
def _blueprint_dir():
|
|
return os.path.join(current_app.instance_path, 'maps')
|
|
|
|
|
|
# =============================================================================
|
|
# Read - public
|
|
# =============================================================================
|
|
|
|
@maplevels_bp.route('', methods=['GET'])
|
|
@require_permission('assets.view')
|
|
def list_levels():
|
|
"""Every active level, grouped by building, in display order.
|
|
|
|
One call, because every consumer needs the whole list: the map switches
|
|
between levels, and a hover preview has to resolve an arbitrary asset's
|
|
level to a blueprint without knowing in advance which one it is.
|
|
"""
|
|
buildings = (Building.query.filter_by(isactive=True)
|
|
.order_by(Building.sortorder, Building.buildingid).all())
|
|
|
|
# How many markers sit on each level, in one grouped query rather than one
|
|
# per level. The admin page needs it to say what a rename or a resize is
|
|
# about to affect, and it is what makes the delete refusal predictable
|
|
# instead of a surprise.
|
|
counts = dict(db.session.query(Asset.levelid, db.func.count(Asset.assetid))
|
|
.filter(Asset.levelid.isnot(None), Asset.isactive.is_(True))
|
|
.group_by(Asset.levelid).all())
|
|
|
|
payload = []
|
|
for building in buildings:
|
|
entry = building.to_dict()
|
|
for level in entry.get('levels', []):
|
|
level['assetcount'] = counts.get(level['levelid'], 0)
|
|
payload.append(entry)
|
|
|
|
default = MapLevel.default_level()
|
|
return success_response({
|
|
'buildings': payload,
|
|
'defaultlevelid': default.levelid if default else None,
|
|
'totalplaced': sum(counts.values()),
|
|
})
|
|
|
|
|
|
@maplevels_bp.route('/<int:levelid>', methods=['GET'])
|
|
@require_permission('assets.view')
|
|
def get_level(levelid):
|
|
level = db.session.get(MapLevel, levelid)
|
|
if not level or not level.isactive:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'No such level',
|
|
http_code=404)
|
|
return success_response(level.to_dict())
|
|
|
|
|
|
@maplevels_bp.route('/<int:levelid>/blueprint/<path:filename>', methods=['GET'])
|
|
@require_permission('assets.view')
|
|
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)
|
|
|
|
|
|
# =============================================================================
|
|
# Write - admin
|
|
# =============================================================================
|
|
|
|
@maplevels_bp.route('/buildings', methods=['POST'])
|
|
@jwt_required()
|
|
@require_role('admin')
|
|
def create_building():
|
|
data = request.get_json() or {}
|
|
name = (data.get('buildingname') or '').strip()
|
|
if not name:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'buildingname is required')
|
|
if Building.query.filter(db.func.lower(Building.buildingname)
|
|
== name.lower()).first():
|
|
return error_response(ErrorCodes.CONFLICT,
|
|
f"Building '{name}' already exists",
|
|
http_code=409)
|
|
building = Building(buildingname=name,
|
|
sortorder=int(data.get('sortorder') or 0))
|
|
db.session.add(building)
|
|
db.session.commit()
|
|
AuditLog.log('created', 'Building', entityid=building.buildingid,
|
|
entityname=name)
|
|
db.session.commit()
|
|
return success_response(building.to_dict(), message='Building created',
|
|
http_code=201)
|
|
|
|
|
|
@maplevels_bp.route('/buildings/<int:buildingid>', methods=['PUT', 'PATCH'])
|
|
@jwt_required()
|
|
@require_role('admin')
|
|
def update_building(buildingid):
|
|
"""Rename or reorder a building. Levels move with it; nothing repositions."""
|
|
building = db.session.get(Building, buildingid)
|
|
if not building:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'No such building',
|
|
http_code=404)
|
|
data = request.get_json() or {}
|
|
if 'buildingname' in data:
|
|
name = (data['buildingname'] or '').strip()
|
|
if not name:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'buildingname cannot be blank')
|
|
clash = Building.query.filter(
|
|
db.func.lower(Building.buildingname) == name.lower(),
|
|
Building.buildingid != buildingid).first()
|
|
if clash:
|
|
return error_response(ErrorCodes.CONFLICT,
|
|
f"Building '{name}' already exists",
|
|
http_code=409)
|
|
building.buildingname = name
|
|
if data.get('sortorder') is not None:
|
|
building.sortorder = int(data['sortorder'])
|
|
if 'isactive' in data:
|
|
building.isactive = bool(data['isactive'])
|
|
db.session.commit()
|
|
AuditLog.log('updated', 'Building', entityid=buildingid,
|
|
entityname=building.buildingname)
|
|
db.session.commit()
|
|
return success_response(building.to_dict(), message='Building updated')
|
|
|
|
|
|
@maplevels_bp.route('', methods=['POST'])
|
|
@jwt_required()
|
|
@require_role('admin')
|
|
def create_level():
|
|
data = request.get_json() or {}
|
|
name = (data.get('levelname') or '').strip()
|
|
buildingid = data.get('buildingid')
|
|
if not name or not buildingid:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'buildingname and buildingid are required')
|
|
if not db.session.get(Building, buildingid):
|
|
return error_response(ErrorCodes.NOT_FOUND, 'No such building',
|
|
http_code=404)
|
|
|
|
level = MapLevel(
|
|
buildingid=buildingid,
|
|
levelname=name,
|
|
sortorder=int(data.get('sortorder') or 0),
|
|
blueprintlight=data.get('blueprintlight'),
|
|
blueprintdark=data.get('blueprintdark'),
|
|
mapwidth=int(data.get('mapwidth') or 3300),
|
|
mapheight=int(data.get('mapheight') or 2550),
|
|
)
|
|
db.session.add(level)
|
|
db.session.flush()
|
|
_apply_default(level, data.get('isdefault'))
|
|
db.session.commit()
|
|
AuditLog.log('created', 'MapLevel', entityid=level.levelid,
|
|
entityname=name)
|
|
db.session.commit()
|
|
return success_response(level.to_dict(), message='Level created',
|
|
http_code=201)
|
|
|
|
|
|
@maplevels_bp.route('/<int:levelid>', methods=['PUT', 'PATCH'])
|
|
@jwt_required()
|
|
@require_role('admin')
|
|
def update_level(levelid):
|
|
level = db.session.get(MapLevel, levelid)
|
|
if not level:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'No such level',
|
|
http_code=404)
|
|
data = request.get_json() or {}
|
|
|
|
# The dimensions are the coordinate space every marker on this level is
|
|
# expressed in, so changing them moves every marker relative to the image.
|
|
# Report it rather than doing it silently; the caller decides whether to run
|
|
# a transform (POST /api/assets/map/transform).
|
|
warnings = []
|
|
for field in ('mapwidth', 'mapheight'):
|
|
if field in data and int(data[field] or 0) != getattr(level, field):
|
|
warnings.append(
|
|
'%s changed from %s to %s; existing positions on this level are '
|
|
'still in the old coordinate space' % (field,
|
|
getattr(level, field),
|
|
data[field]))
|
|
|
|
for field in ('levelname', 'blueprintlight', 'blueprintdark'):
|
|
if field in data:
|
|
setattr(level, field, data[field])
|
|
for field in ('sortorder', 'mapwidth', 'mapheight'):
|
|
if field in data and data[field] is not None:
|
|
setattr(level, field, int(data[field]))
|
|
if 'isactive' in data:
|
|
level.isactive = bool(data['isactive'])
|
|
_apply_default(level, data.get('isdefault'))
|
|
|
|
db.session.commit()
|
|
AuditLog.log('updated', 'MapLevel', entityid=level.levelid,
|
|
entityname=level.levelname)
|
|
db.session.commit()
|
|
payload = level.to_dict()
|
|
if warnings:
|
|
payload['warnings'] = warnings
|
|
return success_response(payload, message='Level updated')
|
|
|
|
|
|
@maplevels_bp.route('/<int:levelid>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_role('admin')
|
|
def delete_level(levelid):
|
|
"""Deactivate a level, refusing while assets are still placed on it.
|
|
|
|
Deleting the drawing out from under a marker would leave a position in a
|
|
coordinate space that no longer exists - unrenderable, and indistinguishable
|
|
from a marker that was never placed.
|
|
"""
|
|
level = db.session.get(MapLevel, levelid)
|
|
if not level:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'No such level',
|
|
http_code=404)
|
|
placed = Asset.query.filter_by(levelid=levelid, isactive=True).count()
|
|
if placed:
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f'{placed} asset(s) are placed on this level. Move them to another '
|
|
f'level first.', http_code=409)
|
|
if level.isdefault:
|
|
return error_response(
|
|
ErrorCodes.VALIDATION_ERROR,
|
|
'This is the default level. Make another level the default first.')
|
|
level.isactive = False
|
|
db.session.commit()
|
|
AuditLog.log('deleted', 'MapLevel', entityid=levelid,
|
|
entityname=level.levelname)
|
|
db.session.commit()
|
|
return success_response(message='Level deactivated')
|
|
|
|
|
|
@maplevels_bp.route('/<int:levelid>/blueprint', methods=['POST'])
|
|
@jwt_required()
|
|
@require_role('admin')
|
|
def upload_blueprint(levelid):
|
|
"""Upload this level's blueprint for one theme.
|
|
|
|
multipart/form-data: file=<image>, theme=light|dark. The native pixel size
|
|
is NOT inferred from the image - it is stated on the level, because that is
|
|
what existing coordinates mean and guessing it would move every marker.
|
|
"""
|
|
level = db.session.get(MapLevel, levelid)
|
|
if not level:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'No such level',
|
|
http_code=404)
|
|
theme = (request.form.get('theme') or '').strip().lower()
|
|
if theme not in ('light', 'dark'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'theme must be light or dark')
|
|
upload = request.files.get('file')
|
|
if not upload or not upload.filename:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No file provided')
|
|
ext = os.path.splitext(upload.filename)[1].lower()
|
|
if ext not in BLUEPRINT_EXTENSIONS:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
f'Unsupported image type {ext}')
|
|
|
|
os.makedirs(_blueprint_dir(), exist_ok=True)
|
|
filename = secure_filename(f'level-{levelid}-{theme}{ext}')
|
|
raw = upload.read()
|
|
upload.seek(0)
|
|
upload.save(os.path.join(_blueprint_dir(), filename))
|
|
url = f'/api/maplevels/{levelid}/blueprint/{filename}'
|
|
setattr(level, f'blueprint{theme}', url)
|
|
|
|
# The image's real pixel size, read from its header. What happens next
|
|
# depends entirely on whether anything is already placed on this level.
|
|
detectedwidth, detectedheight = image_size(raw)
|
|
placed = Asset.query.filter_by(levelid=levelid, isactive=True).count()
|
|
sizenote = None
|
|
if detectedwidth and detectedheight:
|
|
matches = (detectedwidth == level.mapwidth
|
|
and detectedheight == level.mapheight)
|
|
if matches:
|
|
sizenote = None
|
|
elif not placed:
|
|
# Nothing is placed here yet, so no coordinate can be invalidated:
|
|
# adopt the image's own size, which is almost certainly what the
|
|
# operator wanted and saves them typing it.
|
|
level.mapwidth = detectedwidth
|
|
level.mapheight = detectedheight
|
|
sizenote = ('dimensions set from the image: %d x %d'
|
|
% (detectedwidth, detectedheight))
|
|
else:
|
|
# Markers exist in the OLD coordinate space. Silently adopting the
|
|
# new size would move every one of them relative to the drawing
|
|
# while looking like a successful upload, so this reports and
|
|
# changes nothing. Resizing is a transform, not an upload.
|
|
sizenote = (
|
|
'this image is %d x %d but the level is set to %d x %d, and %d '
|
|
'marker(s) are placed in the current space. The dimensions were '
|
|
'NOT changed: set them with a landmark transform '
|
|
'(POST /api/mappositions/transform) so the markers move with '
|
|
'them.' % (detectedwidth, detectedheight, level.mapwidth,
|
|
level.mapheight, placed))
|
|
|
|
db.session.commit()
|
|
AuditLog.log('updated', 'MapLevel', entityid=levelid,
|
|
entityname=level.levelname,
|
|
changes={f'blueprint{theme}': {'new': url}})
|
|
db.session.commit()
|
|
payload = {'levelid': levelid, f'blueprint{theme}': url,
|
|
'mapwidth': level.mapwidth, 'mapheight': level.mapheight,
|
|
'detectedwidth': detectedwidth, 'detectedheight': detectedheight,
|
|
'placedassets': placed}
|
|
if sizenote:
|
|
payload['sizenote'] = sizenote
|
|
return success_response(payload, message='Blueprint uploaded')
|
|
|
|
|
|
def _apply_default(level, requested):
|
|
"""Make this level the default, clearing the flag elsewhere.
|
|
|
|
Exactly-one-default is a rule across rows, which no column constraint can
|
|
express, so it is enforced here - the one place that sets the flag.
|
|
"""
|
|
if not requested:
|
|
return
|
|
MapLevel.query.filter(MapLevel.levelid != level.levelid).update(
|
|
{'isdefault': False})
|
|
level.isdefault = True
|