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.
243 lines
8.1 KiB
Python
243 lines
8.1 KiB
Python
"""Locations API endpoints - Full CRUD."""
|
|
|
|
from flask import Blueprint, request
|
|
from flask_jwt_extended import jwt_required
|
|
|
|
from shopdb.extensions import db
|
|
from shopdb.core.models import Location, LocationType
|
|
from shopdb.utils.responses import (
|
|
success_response,
|
|
error_response,
|
|
paginated_response,
|
|
ErrorCodes
|
|
)
|
|
from shopdb.utils.pagination import get_pagination_params, paginate_query
|
|
|
|
from shopdb.utils.authz import require_role
|
|
from shopdb.utils.import_mode import apply_import_timestamps
|
|
from shopdb.utils.authz import require_permission
|
|
|
|
locations_bp = Blueprint('locations', __name__)
|
|
|
|
|
|
def _loc_type_dict(t):
|
|
return {
|
|
'locationtypeid': t.locationtypeid,
|
|
'locationtype': t.locationtype,
|
|
'description': t.description,
|
|
'color': t.color,
|
|
'isactive': t.isactive,
|
|
}
|
|
|
|
|
|
@locations_bp.route('/types', methods=['GET'])
|
|
@require_permission('assets.view')
|
|
def list_location_types():
|
|
"""List location types. ?active=false includes inactive ones."""
|
|
query = LocationType.query
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter_by(isactive=True)
|
|
types = query.order_by(LocationType.locationtype).all()
|
|
return success_response([_loc_type_dict(t) for t in types])
|
|
|
|
|
|
@locations_bp.route('/types', methods=['POST'])
|
|
@jwt_required()
|
|
@require_role('admin')
|
|
def create_location_type():
|
|
data = request.get_json() or {}
|
|
if not data.get('locationtype'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'locationtype is required')
|
|
existing = LocationType.query.filter_by(locationtype=data['locationtype']).first()
|
|
if existing:
|
|
if not existing.isactive:
|
|
existing.isactive = True
|
|
for key in ('description', 'color'):
|
|
if data.get(key) is not None:
|
|
setattr(existing, key, data[key])
|
|
db.session.commit()
|
|
return success_response(_loc_type_dict(existing), message='Reactivated existing type')
|
|
return error_response(ErrorCodes.CONFLICT,
|
|
f"Location type '{data['locationtype']}' already exists", http_code=409)
|
|
t = LocationType(locationtype=data['locationtype'],
|
|
description=data.get('description'), color=data.get('color'))
|
|
db.session.add(t)
|
|
db.session.commit()
|
|
return success_response(_loc_type_dict(t), message='Location type created', http_code=201)
|
|
|
|
|
|
@locations_bp.route('/types/<int:type_id>', methods=['PUT'])
|
|
@jwt_required()
|
|
@require_role('admin')
|
|
def update_location_type(type_id):
|
|
t = db.session.get(LocationType, type_id)
|
|
if not t:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Location type not found', http_code=404)
|
|
data = request.get_json() or {}
|
|
if 'locationtype' in data and data['locationtype'] != t.locationtype:
|
|
if LocationType.query.filter_by(locationtype=data['locationtype']).first():
|
|
return error_response(ErrorCodes.CONFLICT,
|
|
f"Location type '{data['locationtype']}' already exists", http_code=409)
|
|
for key in ('locationtype', 'description', 'color', 'isactive'):
|
|
if key in data:
|
|
setattr(t, key, data[key])
|
|
db.session.commit()
|
|
return success_response(_loc_type_dict(t), message='Location type updated')
|
|
|
|
|
|
@locations_bp.route('/types/<int:type_id>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_role('admin')
|
|
def delete_location_type(type_id):
|
|
t = db.session.get(LocationType, type_id)
|
|
if not t:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Location type not found', http_code=404)
|
|
inuse = Location.query.filter_by(locationtypeid=type_id).count()
|
|
if inuse:
|
|
return error_response(ErrorCodes.CONFLICT,
|
|
f"Cannot delete: {inuse} location(s) still use this type", http_code=409)
|
|
db.session.delete(t)
|
|
db.session.commit()
|
|
return success_response(message='Location type deleted')
|
|
|
|
|
|
@locations_bp.route('', methods=['GET'])
|
|
@require_permission('assets.view')
|
|
def list_locations():
|
|
"""List all locations."""
|
|
page, per_page = get_pagination_params(request)
|
|
|
|
query = Location.query
|
|
|
|
if request.args.get('active', 'true').lower() != 'false':
|
|
query = query.filter(Location.isactive == True)
|
|
|
|
# Exact-match natural-key lookup for idempotent import (location name).
|
|
if exactname := request.args.get('locationname'):
|
|
query = query.filter(Location.locationname == exactname)
|
|
|
|
if search := request.args.get('search'):
|
|
query = query.filter(
|
|
db.or_(
|
|
Location.locationname.ilike(f'%{search}%'),
|
|
Location.building.ilike(f'%{search}%')
|
|
)
|
|
)
|
|
|
|
query = query.order_by(Location.locationname)
|
|
|
|
items, total = paginate_query(query, page, per_page)
|
|
data = [loc.to_dict() for loc in items]
|
|
|
|
return paginated_response(data, page, per_page, total)
|
|
|
|
|
|
@locations_bp.route('/<int:location_id>', methods=['GET'])
|
|
@require_permission('assets.view')
|
|
def get_location(location_id: int):
|
|
"""Get a single location."""
|
|
loc = db.session.get(Location, location_id)
|
|
|
|
if not loc:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Location with ID {location_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
return success_response(loc.to_dict())
|
|
|
|
|
|
@locations_bp.route('', methods=['POST'])
|
|
@jwt_required()
|
|
@require_role('admin')
|
|
def create_location():
|
|
"""Create a new location."""
|
|
data = request.get_json()
|
|
|
|
if not data or not data.get('locationname'):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'locationname is required')
|
|
|
|
if Location.query.filter_by(locationname=data['locationname']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Location '{data['locationname']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
loc = Location(
|
|
locationname=data['locationname'],
|
|
building=data.get('building'),
|
|
floor=data.get('floor'),
|
|
room=data.get('room'),
|
|
description=data.get('description'),
|
|
locationtypeid=data.get('locationtypeid'),
|
|
parentlocationid=data.get('parentlocationid'),
|
|
mapimage=data.get('mapimage'),
|
|
mapwidth=data.get('mapwidth'),
|
|
mapheight=data.get('mapheight')
|
|
)
|
|
|
|
db.session.add(loc)
|
|
apply_import_timestamps(loc, data)
|
|
db.session.commit()
|
|
|
|
return success_response(loc.to_dict(), message='Location created', http_code=201)
|
|
|
|
|
|
@locations_bp.route('/<int:location_id>', methods=['PUT'])
|
|
@jwt_required()
|
|
@require_role('admin')
|
|
def update_location(location_id: int):
|
|
"""Update a location."""
|
|
loc = db.session.get(Location, location_id)
|
|
|
|
if not loc:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Location with ID {location_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
data = request.get_json()
|
|
if not data:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided')
|
|
|
|
if 'locationname' in data and data['locationname'] != loc.locationname:
|
|
if Location.query.filter_by(locationname=data['locationname']).first():
|
|
return error_response(
|
|
ErrorCodes.CONFLICT,
|
|
f"Location '{data['locationname']}' already exists",
|
|
http_code=409
|
|
)
|
|
|
|
for key in ['locationname', 'building', 'floor', 'room', 'description',
|
|
'locationtypeid', 'parentlocationid', 'mapimage', 'mapwidth',
|
|
'mapheight', 'isactive']:
|
|
if key in data:
|
|
setattr(loc, key, data[key])
|
|
|
|
apply_import_timestamps(loc, data)
|
|
db.session.commit()
|
|
return success_response(loc.to_dict(), message='Location updated')
|
|
|
|
|
|
@locations_bp.route('/<int:location_id>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_role('admin')
|
|
def delete_location(location_id: int):
|
|
"""Delete (deactivate) a location."""
|
|
loc = db.session.get(Location, location_id)
|
|
|
|
if not loc:
|
|
return error_response(
|
|
ErrorCodes.NOT_FOUND,
|
|
f'Location with ID {location_id} not found',
|
|
http_code=404
|
|
)
|
|
|
|
loc.isactive = False
|
|
db.session.commit()
|
|
|
|
return success_response(message='Location deleted')
|