ced356882c
IIS ARR sets X-Forwarded-For to clientip:port, and the port changes every connection. Left in, the audit log showed IP:PORT, the dashboard IP fallback never matched a stored (portless) DashboardDefault.ipaddress, and login rate limiting keyed per-connection instead of per-host. Add an IPv6-safe clientip.client_ip / strip_port helper and use it in the audit log, the dashboard resolver, and the login rate-limit key.
212 lines
8.3 KiB
Python
212 lines
8.3 KiB
Python
"""Dashboard defaults API: visitor-IP -> business-unit mapping.
|
|
|
|
CRUD for the mappings plus a resolve endpoint a kiosk/lobby dashboard calls to
|
|
auto-select its business unit from the display PC's IP.
|
|
"""
|
|
|
|
from flask import Blueprint, request
|
|
from flask_jwt_extended import jwt_required
|
|
|
|
from shopdb.extensions import db
|
|
from shopdb.core.models import DashboardDefault, BusinessUnit, AuditLog
|
|
from shopdb.core.models.dashboarddefault import DISPLAY_ROLES, DISPLAY_ROLE_PATHS
|
|
from shopdb.utils.responses import success_response, error_response, ErrorCodes
|
|
from shopdb.utils.authz import require_role
|
|
|
|
dashboarddefaults_bp = Blueprint('dashboarddefaults', __name__)
|
|
|
|
|
|
def _request_ip():
|
|
"""Caller IP for the IP fallback, port stripped so it matches a stored
|
|
(portless) DashboardDefault.ipaddress. ARR forwards clientip:port."""
|
|
from shopdb.utils.clientip import client_ip
|
|
return client_ip(request)
|
|
|
|
|
|
def _serialize(default):
|
|
data = default.to_dict()
|
|
data['businessunit'] = default.businessunit.businessunit \
|
|
if default.businessunit else None
|
|
data['displaypath'] = default.displaypath
|
|
return data
|
|
|
|
|
|
def _resolve_default(fqdn, ipaddress):
|
|
"""Find the active mapping for a display: FQDN first (stable), then IP."""
|
|
default = None
|
|
if fqdn:
|
|
default = DashboardDefault.query.filter_by(
|
|
fqdn=fqdn, isactive=True).first()
|
|
if not default and ipaddress:
|
|
default = DashboardDefault.query.filter_by(
|
|
ipaddress=ipaddress, isactive=True).first()
|
|
return default
|
|
|
|
|
|
@dashboarddefaults_bp.route('/visitor-location', methods=['GET'])
|
|
def visitor_location():
|
|
"""Resolve the business unit for the calling display by its IP.
|
|
|
|
Unauthenticated: kiosks/lobby displays hit this. Returns the mapped
|
|
business unit, or a null businessunitid when the IP is not mapped.
|
|
"""
|
|
fqdn = (request.args.get('fqdn') or '').strip().lower() or None
|
|
ipaddress = request.args.get('ipaddress') or _request_ip()
|
|
default = _resolve_default(fqdn, ipaddress)
|
|
if not default:
|
|
return success_response({
|
|
'fqdn': fqdn,
|
|
'ipaddress': ipaddress,
|
|
'businessunitid': None,
|
|
'businessunit': None,
|
|
})
|
|
return success_response({
|
|
'fqdn': fqdn,
|
|
'ipaddress': ipaddress,
|
|
'businessunitid': default.businessunitid,
|
|
'businessunit': default.businessunit.businessunit
|
|
if default.businessunit else None,
|
|
})
|
|
|
|
|
|
@dashboarddefaults_bp.route('/display-role', methods=['GET'])
|
|
def display_role():
|
|
"""Resolve what a single 'display' PC should show, from its own IP.
|
|
|
|
Unauthenticated: the display launcher calls this at boot. Returns the role
|
|
(dashboard / lobby / partskiosk), the frontend path it maps to, and the
|
|
business unit for the dashboard role. Unmapped IP -> null role.
|
|
"""
|
|
fqdn = (request.args.get('fqdn') or '').strip().lower() or None
|
|
ipaddress = request.args.get('ipaddress') or _request_ip()
|
|
default = _resolve_default(fqdn, ipaddress)
|
|
if not default:
|
|
return success_response({
|
|
'fqdn': fqdn,
|
|
'ipaddress': ipaddress,
|
|
'role': None,
|
|
'path': None,
|
|
'businessunitid': None,
|
|
'businessunit': None,
|
|
})
|
|
return success_response({
|
|
'fqdn': fqdn,
|
|
'ipaddress': ipaddress,
|
|
'role': default.displayrole,
|
|
'path': default.displaypath,
|
|
'businessunitid': default.businessunitid,
|
|
'businessunit': default.businessunit.businessunit
|
|
if default.businessunit else None,
|
|
})
|
|
|
|
|
|
@dashboarddefaults_bp.route('', methods=['GET'])
|
|
@jwt_required(optional=True)
|
|
def list_defaults():
|
|
"""List all visitor-IP -> business-unit mappings."""
|
|
defaults = DashboardDefault.query.filter_by(isactive=True).order_by(
|
|
DashboardDefault.ipaddress).all()
|
|
return success_response([_serialize(d) for d in defaults])
|
|
|
|
|
|
@dashboarddefaults_bp.route('', methods=['POST'])
|
|
@jwt_required()
|
|
@require_role('admin')
|
|
def create_default():
|
|
"""Create a visitor-IP -> business-unit mapping."""
|
|
data = request.get_json() or {}
|
|
fqdn = (data.get('fqdn') or '').strip().lower() or None
|
|
ipaddress = (data.get('ipaddress') or '').strip() or None
|
|
role = (data.get('displayrole') or 'dashboard').strip()
|
|
businessunitid = data.get('businessunitid')
|
|
|
|
if not fqdn and not ipaddress:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'fqdn or ipaddress is required')
|
|
if role not in DISPLAY_ROLES:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
f'displayrole must be one of {", ".join(DISPLAY_ROLES)}')
|
|
# Only the dashboard role needs a business unit (location).
|
|
if role == 'dashboard':
|
|
if not businessunitid:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'a location is required for the dashboard role')
|
|
if not db.session.get(BusinessUnit, businessunitid):
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Location not found',
|
|
http_code=404)
|
|
else:
|
|
businessunitid = None
|
|
if fqdn and DashboardDefault.query.filter_by(fqdn=fqdn, isactive=True).first():
|
|
return error_response(ErrorCodes.CONFLICT,
|
|
f"{fqdn} is already mapped", http_code=409)
|
|
if ipaddress and DashboardDefault.query.filter_by(
|
|
ipaddress=ipaddress, isactive=True).first():
|
|
return error_response(ErrorCodes.CONFLICT,
|
|
f"IP {ipaddress} is already mapped", http_code=409)
|
|
|
|
default = DashboardDefault(fqdn=fqdn, ipaddress=ipaddress, displayrole=role,
|
|
businessunitid=businessunitid,
|
|
description=data.get('description'))
|
|
db.session.add(default)
|
|
db.session.flush()
|
|
AuditLog.log('created', 'DashboardDefault', entityid=default.dashboarddefaultid,
|
|
entityname=(fqdn or ipaddress))
|
|
db.session.commit()
|
|
return success_response(_serialize(default), message='Mapping created',
|
|
http_code=201)
|
|
|
|
|
|
@dashboarddefaults_bp.route('/<int:default_id>', methods=['PUT'])
|
|
@jwt_required()
|
|
@require_role('admin')
|
|
def update_default(default_id):
|
|
"""Update a mapping."""
|
|
default = db.session.get(DashboardDefault, default_id)
|
|
if not default:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Mapping not found', http_code=404)
|
|
|
|
data = request.get_json() or {}
|
|
if 'displayrole' in data:
|
|
role = (data.get('displayrole') or '').strip()
|
|
if role not in DISPLAY_ROLES:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
f'displayrole must be one of {", ".join(DISPLAY_ROLES)}')
|
|
default.displayrole = role
|
|
if 'businessunitid' in data:
|
|
if data['businessunitid'] and not db.session.get(BusinessUnit, data['businessunitid']):
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Business unit not found',
|
|
http_code=404)
|
|
default.businessunitid = data['businessunitid']
|
|
if 'fqdn' in data:
|
|
default.fqdn = (data.get('fqdn') or '').strip().lower() or None
|
|
if 'ipaddress' in data:
|
|
default.ipaddress = (data.get('ipaddress') or '').strip() or None
|
|
if 'description' in data:
|
|
default.description = data['description']
|
|
|
|
if not default.fqdn and not default.ipaddress:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'fqdn or ipaddress is required')
|
|
# Non-dashboard roles carry no location; dashboard needs one.
|
|
if default.displayrole != 'dashboard':
|
|
default.businessunitid = None
|
|
elif not default.businessunitid:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR,
|
|
'a location is required for the dashboard role')
|
|
|
|
db.session.commit()
|
|
return success_response(_serialize(default), message='Mapping updated')
|
|
|
|
|
|
@dashboarddefaults_bp.route('/<int:default_id>', methods=['DELETE'])
|
|
@jwt_required()
|
|
@require_role('admin')
|
|
def delete_default(default_id):
|
|
"""Delete (deactivate) a mapping."""
|
|
default = db.session.get(DashboardDefault, default_id)
|
|
if not default:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Mapping not found', http_code=404)
|
|
default.isactive = False
|
|
db.session.commit()
|
|
return success_response(message='Mapping deleted')
|