"""Agent binary upload and retention routes. Lets an operator drop in a new shopdbagent.exe or shopdb-endpoint-tray.exe from the Endpoints page, keep the last few versions, and roll the current pointer back to a retained build. The actual staging/retention logic lives in agentstore.py; this module is only the HTTP surface and authorization. Authorization boundary (deliberate): - Uploading, listing, and selecting the current staged version is endpoints.agentpackage - the same grant that already lets someone build and download the site-agent ZIP. Staging a binary only changes what a FUTURE download contains; it does not touch any running endpoint. - Pushing that binary to live endpoints (agent replacement) is a separate, heavier grant, endpoints.agentupdate, and is NOT wired here. It is a distinct command family from application deploys (endpoints.deploy) so app-push rights never imply SYSTEM-level agent replacement. That push pipe is a later slice. Uploads are capped and PE-checked so a wrong or hostile file cannot be staged. """ from flask import request from shopdb.api import (success_response, error_response, ErrorCodes, require_permission) from .routes import endpoints_bp from . import agentstore # Hard cap on an uploaded agent binary. The real agent is a few MB; this is a # generous ceiling that still refuses an operator who points at a giant wrong # file before it is buffered fully into memory-staged storage. _MAX_UPLOAD_BYTES = 64 * 1024 * 1024 @endpoints_bp.route('/siteagent/binaries/', methods=['GET']) @require_permission('endpoints.agentpackage') def list_agent_binaries(binkey): """List staged versions for a binary (agent or tray), newest first.""" if not agentstore.is_valid_binary_key(binkey): return error_response( ErrorCodes.VALIDATION_ERROR, 'binkey must be one of: agent, tray', http_code=400) return success_response({ 'binary': binkey, 'current': agentstore.current_version(binkey), 'versions': agentstore.list_versions(binkey), }) @endpoints_bp.route('/siteagent/binaries/', methods=['POST']) @require_permission('endpoints.agentpackage') def upload_agent_binary(binkey): """Stage an uploaded binary as a new version. Expects multipart/form-data with a 'file' upload and a 'version' field. The version must be a strict dotted-numeric string (for example 1.6.3) so it is safe as a directory name and orderable. The upload must be a Windows PE executable. Older versions past the retention window are pruned, but never the current one. By default the upload becomes the current staged build; send makecurrent=false to stage a rollback/legacy build that joins the version list without displacing the active current. """ version = (request.form.get('version') or '').strip() makecurrent = (request.form.get('makecurrent') or 'true').strip().lower() makecurrent = makecurrent not in ('false', '0', 'no') if not agentstore.is_valid_binary_key(binkey): return error_response( ErrorCodes.VALIDATION_ERROR, 'binkey must be one of: agent, tray', http_code=400) if not agentstore.is_valid_version(version): return error_response( ErrorCodes.VALIDATION_ERROR, 'version must be a dotted-numeric string such as 1.6.3', http_code=400) upload = request.files.get('file') if upload is None or not upload.filename: return error_response( ErrorCodes.VALIDATION_ERROR, 'a file upload is required', http_code=400) data = upload.read(_MAX_UPLOAD_BYTES + 1) if len(data) > _MAX_UPLOAD_BYTES: return error_response( ErrorCodes.VALIDATION_ERROR, 'uploaded binary exceeds the size limit', http_code=400) if not agentstore.looks_like_pe(data): return error_response( ErrorCodes.VALIDATION_ERROR, 'uploaded file is not a Windows executable', http_code=400) result = agentstore.store_version(binkey, version, data, make_current=makecurrent) return success_response({ 'binary': binkey, 'current': result['current'], 'versions': result['versions'], }) @endpoints_bp.route('/siteagent/binaries//detect', methods=['POST']) @require_permission('endpoints.agentpackage') def detect_agent_binary_version(binkey): """Read the PE file version of an uploaded EXE without staging it. The Build Site Agent page calls this on file-select so the version field is prefilled from the binary's own version resource - the operator does not have to read the EXE properties by hand for a current-release upload. Returns {version} which is null when the file carries no readable version. """ if not agentstore.is_valid_binary_key(binkey): return error_response( ErrorCodes.VALIDATION_ERROR, 'binkey must be one of: agent, tray', http_code=400) upload = request.files.get('file') if upload is None or not upload.filename: return error_response( ErrorCodes.VALIDATION_ERROR, 'a file upload is required', http_code=400) data = upload.read(_MAX_UPLOAD_BYTES + 1) if len(data) > _MAX_UPLOAD_BYTES: return error_response( ErrorCodes.VALIDATION_ERROR, 'uploaded binary exceeds the size limit', http_code=400) if not agentstore.looks_like_pe(data): return error_response( ErrorCodes.VALIDATION_ERROR, 'uploaded file is not a Windows executable', http_code=400) return success_response({ 'binary': binkey, 'version': agentstore.read_pe_file_version(data), }) @endpoints_bp.route('/siteagent/binaries//adopt', methods=['POST']) @require_permission('endpoints.agentpackage') def adopt_legacy_agent_binary(binkey): """Adopt the unmanaged legacy flat file into the version store. The pre-versioning install dropped a single shopdbagent.exe (or tray exe) straight under siteagent/. This copies that file in as a real staged version so it shows a version and joins retention, without the operator having to re-upload it. The version comes from the request when the binary carries no readable PE version resource; otherwise the detected version is used. """ if not agentstore.is_valid_binary_key(binkey): return error_response( ErrorCodes.VALIDATION_ERROR, 'binkey must be one of: agent, tray', http_code=400) status = agentstore.legacy_status(binkey) if status is None: return error_response( ErrorCodes.NOT_FOUND, 'there is no unmanaged legacy binary to adopt', http_code=404) source = request.get_json(silent=True) or {} version = (source.get('version') or status.get('version') or '').strip() if not agentstore.is_valid_version(version): return error_response( ErrorCodes.VALIDATION_ERROR, 'version must be a dotted-numeric string such as 1.6.3', http_code=400) result = agentstore.adopt_legacy(binkey, version) if result is None: return error_response( ErrorCodes.VALIDATION_ERROR, 'the legacy binary could not be adopted', http_code=400) return success_response({ 'binary': binkey, 'current': result['current'], 'versions': result['versions'], }) @endpoints_bp.route('/siteagent/binaries//current', methods=['PUT']) @require_permission('endpoints.agentpackage') def set_current_agent_binary(binkey): """Point 'current' at an already-staged version (roll back or forward).""" if not agentstore.is_valid_binary_key(binkey): return error_response( ErrorCodes.VALIDATION_ERROR, 'binkey must be one of: agent, tray', http_code=400) source = request.get_json(silent=True) or {} version = (source.get('version') or '').strip() if not agentstore.is_valid_version(version): return error_response( ErrorCodes.VALIDATION_ERROR, 'version must be a dotted-numeric string such as 1.6.3', http_code=400) if not agentstore.set_current(binkey, version): return error_response( ErrorCodes.NOT_FOUND, 'that version is not staged', http_code=404) return success_response({ 'binary': binkey, 'current': agentstore.current_version(binkey), 'versions': agentstore.list_versions(binkey), })