"""Applications API endpoints.""" import glob import hashlib import re 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 ( Application, AppVersion, AuditLog ) 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.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 # site's files are not mixed into the code tree and survive a redeploy. APP_IMAGE_EXTENSIONS = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.svg', '.ico'} APP_IMAGE_URL_PREFIX = '/api/applications/image/' # Installers are whatever a vendor ships. The list is a guard against a browser # handing us something unintended, not a security boundary - the download route # always sends as an attachment and never executes anything. APP_PACKAGE_EXTENSIONS = {'.exe', '.msi', '.msp', '.zip', '.7z', '.cab', '.iso', '.appx', '.msix', '.ps1', '.bat', '.txt', '.pdf'} APP_PACKAGE_URL_PREFIX = '/api/applications/package/' # A real installer runs to hundreds of megabytes; an ISO runs to gigabytes and # does not belong in an instance directory. 500MB is the line: enough for the # msi/exe this is for, small enough that the disk cannot vanish behind it. MAX_PACKAGE_BYTES = 500 * 1024 * 1024 def _appimage_dir(): return os.path.join(current_app.instance_path, 'applicationimages') def _apppackage_dir(): return os.path.join(current_app.instance_path, 'applicationpackages') def _sha256_of(stream): """Hex SHA-256 of an upload stream, read in chunks, cursor left rewound. The agent pins a version to exact bytes; computing the digest here means the server is the one source of truth for what those bytes are. """ digest = hashlib.sha256() stream.seek(0) for chunk in iter(lambda: stream.read(1048576), b''): digest.update(chunk) stream.seek(0) return digest.hexdigest() def _replace_upload(directory, stem, ext, upload): """Save one file per application, replacing any prior extension. Returns the stored filename. Without the glob a re-upload as .png would orphan the old .jpg and the two would fight over which is current. """ os.makedirs(directory, exist_ok=True) for old in glob.glob(os.path.join(directory, secure_filename(stem) + '.*')): os.remove(old) filename = secure_filename(f'{stem}{ext}') upload.save(os.path.join(directory, filename)) return filename def _computer_models(): """Lazily import the computers plugin models, or None if unavailable. Application install-tracking is a join over the computers plugin's tables. Importing lazily keeps the applications API importable when the computers plugin is absent or disabled. """ try: from plugins.computers.models import Computer, ComputerInstalledApp return Computer, ComputerInstalledApp except ImportError: return None def _installed_count(appid): """Count active installs of an app, 0 when the computers plugin is absent.""" models = _computer_models() if not models: return 0 _, ComputerInstalledApp = models return ComputerInstalledApp.query.filter_by(appid=appid, isactive=True).count() def _related_kb(appid): """Knowledge Base articles linked to this app, [] when the KB plugin is absent. Lazy + guarded so core stays decoupled from the plugin.""" try: from plugins.knowledgebase.models import KnowledgeBase except ImportError: return [] rows = KnowledgeBase.query.filter_by(appid=appid, isactive=True).order_by( KnowledgeBase.shortdescription).all() return [{'linkid': k.linkid, 'shortdescription': k.shortdescription, 'linkurl': k.linkurl, 'keywords': k.keywords} for k in rows] def _require_computer_models(): """Resolve (Computer, ComputerInstalledApp) or a 503 response tuple. Usage: `models, err = _require_computer_models(); if err: return err`. """ models = _computer_models() if not models: return None, error_response( ErrorCodes.INTERNAL_ERROR, 'Install tracking requires the computers plugin', http_code=503) return models, None from shopdb.utils.authz import require_permission from shopdb.utils.uploads import send_upload applications_bp = Blueprint('applications', __name__) @applications_bp.route('', methods=['GET']) @require_permission('applications.view') def list_applications(): """List all applications.""" page, per_page = get_pagination_params(request) query = Application.query if request.args.get('active', 'true').lower() != 'false': query = query.filter(Application.isactive == True) # Filter out hidden unless specifically requested if request.args.get('showhidden', 'false').lower() != 'true': query = query.filter(Application.ishidden == False) # Filter by installable if request.args.get('installable') is not None: installable = request.args.get('installable').lower() == 'true' query = query.filter(Application.isinstallable == installable) # Exact-match natural-key lookup for idempotent import (app name). if exactappname := request.args.get('appname'): query = query.filter(Application.appname == exactappname) if search := request.args.get('search'): query = query.filter( db.or_( Application.appname.ilike(f'%{search}%'), Application.appdescription.ilike(f'%{search}%') ) ) query = query.order_by(Application.appname) items, total = paginate_query(query, page, per_page) data = [] for app in items: # to_dict already flattens supportteamname/teamurl/contacts. app_dict = app.to_dict() app_dict['installedcount'] = _installed_count(app.appid) data.append(app_dict) return paginated_response(data, page, per_page, total) @applications_bp.route('/', methods=['GET']) @require_permission('applications.view') def get_application(app_id: int): """Get a single application with details.""" app = db.session.get(Application, app_id) if not app: return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404) data = app.to_dict() data['versions'] = [v.to_dict() for v in app.versions.filter_by(isactive=True).order_by(AppVersion.version.desc()).all()] data['installedcount'] = _installed_count(app.appid) data['knowledgebase'] = _related_kb(app.appid) return success_response(data) @applications_bp.route('', methods=['POST']) @require_permission('applications.create') def create_application(): """Create a new application.""" data = request.get_json() if not data or not data.get('appname'): return error_response(ErrorCodes.VALIDATION_ERROR, 'appname is required') if Application.query.filter_by(appname=data['appname']).first(): return error_response( ErrorCodes.CONFLICT, f"Application '{data['appname']}' already exists", http_code=409 ) app = Application( appname=data['appname'], appdescription=data.get('appdescription'), supportteamid=data.get('supportteamid'), isinstallable=data.get('isinstallable', False), applicationnotes=data.get('applicationnotes'), installpath=data.get('installpath'), applicationlink=data.get('applicationlink'), documentationpath=data.get('documentationpath'), ishidden=data.get('ishidden', False), isprinter=data.get('isprinter', False), islicenced=data.get('islicenced', False), isrequired=data.get('isrequired', False), image=data.get('image') ) db.session.add(app) apply_import_timestamps(app, data) db.session.flush() AuditLog.log('created', 'Application', entityid=app.appid, entityname=app.appname) db.session.commit() return success_response(app.to_dict(), message='Application created', http_code=201) @applications_bp.route('/', methods=['PUT']) @require_permission('applications.edit') def update_application(app_id: int): """Update an application.""" app = db.session.get(Application, app_id) if not app: return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404) data = request.get_json() if not data: return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided') if 'appname' in data and data['appname'] != app.appname: if Application.query.filter_by(appname=data['appname']).first(): return error_response( ErrorCodes.CONFLICT, f"Application '{data['appname']}' already exists", http_code=409 ) fields = [ 'appname', 'appdescription', 'supportteamid', 'isinstallable', 'applicationnotes', 'installpath', 'applicationlink', 'documentationpath', 'ishidden', 'isprinter', 'islicenced', 'isrequired', 'image', 'isactive' ] changes = {} for key in fields: if key in data: old_val = getattr(app, key) new_val = data[key] if old_val != new_val: changes[key] = {'old': old_val, 'new': new_val} setattr(app, key, data[key]) if changes: AuditLog.log('updated', 'Application', entityid=app.appid, entityname=app.appname, changes=changes) apply_import_timestamps(app, data) db.session.commit() return success_response(app.to_dict(), message='Application updated') @applications_bp.route('/', methods=['DELETE']) @require_permission('applications.delete') def delete_application(app_id: int): """Delete (deactivate) an application.""" app = db.session.get(Application, app_id) if not app: return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404) app.isactive = False AuditLog.log('deleted', 'Application', entityid=app.appid, entityname=app.appname) db.session.commit() return success_response(message='Application deleted') # ---- Versions ---- @applications_bp.route('//versions', methods=['GET']) @require_permission('applications.view') def list_versions(app_id: int): """List all versions for an application.""" app = db.session.get(Application, app_id) if not app: return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404) versions = app.versions.filter_by(isactive=True).order_by(AppVersion.version.desc()).all() return success_response([v.to_dict() for v in versions]) @applications_bp.route('//versions', methods=['POST']) @require_permission('applications.create') def create_version(app_id: int): """Create a new version for an application.""" app = db.session.get(Application, app_id) if not app: return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404) data = request.get_json() 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, f"Version '{data['version']}' already exists for this application", http_code=409 ) version = AppVersion( appid=app_id, version=data['version'], releasedate=releasedate, notes=data.get('notes'), sha256=digest or None, silentargs=data.get('silentargs') ) db.session.add(version) # Preserve the legacy release/added timestamps in import mode (no-op # otherwise), so imported version history keeps its original dates. apply_import_timestamps(version, data) db.session.commit() return success_response(version.to_dict(), message='Version created', http_code=201) # ---- Computers with this app installed ---- @applications_bp.route('//installed', methods=['GET']) @require_permission('applications.view') def list_installed_machines(app_id: int): """List all computers that have this application installed.""" app = db.session.get(Application, app_id) if not app: return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404) models, err = _require_computer_models() if err: return err _, ComputerInstalledApp = models installed = ComputerInstalledApp.query.filter_by( appid=app_id, isactive=True).all() data = [] for i in installed: comp = i.computer version = i.installedversion if not version and i.appversion: version = i.appversion.version item = { 'id': i.id, 'computerid': i.computerid, 'version': version, } if comp: item['computer'] = { 'computerid': comp.computerid, 'assetnumber': comp.asset.assetnumber if comp.asset else None, 'hostname': comp.hostname, } data.append(item) return success_response(data) # ---- Installed Apps (per computer) ---- @applications_bp.route('/machines/', methods=['GET']) @require_permission('applications.view') def list_machine_applications(machine_id: int): """List all applications installed on a computer.""" models, err = _require_computer_models() if err: return err Computer, ComputerInstalledApp = models comp = db.session.get(Computer, machine_id) if not comp: return error_response(ErrorCodes.NOT_FOUND, 'Computer not found', http_code=404) installed = comp.installedapps.filter_by(isactive=True).all() return success_response([i.to_dict() for i in installed]) @applications_bp.route('/machines/', methods=['POST']) @require_permission('applications.create') def install_application(machine_id: int): """Install an application on a computer.""" models, err = _require_computer_models() if err: return err Computer, ComputerInstalledApp = models comp = db.session.get(Computer, machine_id) if not comp: return error_response(ErrorCodes.NOT_FOUND, 'Computer not found', http_code=404) data = request.get_json() if not data or not data.get('appid'): return error_response(ErrorCodes.VALIDATION_ERROR, 'appid is required') app = db.session.get(Application, data['appid']) if not app: return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404) existing = ComputerInstalledApp.query.filter_by( computerid=machine_id, appid=data['appid'] ).first() if existing: if existing.isactive: return error_response( ErrorCodes.CONFLICT, 'Application already installed on this computer', http_code=409 ) existing.isactive = True existing.appversionid = data.get('appversionid') existing.installeddate = db.func.now() db.session.commit() return success_response(existing.to_dict(), message='Application reinstalled') installed = ComputerInstalledApp( computerid=machine_id, appid=data['appid'], appversionid=data.get('appversionid') ) db.session.add(installed) db.session.commit() return success_response(installed.to_dict(), message='Application installed', http_code=201) @applications_bp.route('/machines//', methods=['DELETE']) @require_permission('applications.delete') def uninstall_application(machine_id: int, app_id: int): """Uninstall an application from a computer.""" models, err = _require_computer_models() if err: return err _, ComputerInstalledApp = models installed = ComputerInstalledApp.query.filter_by( computerid=machine_id, appid=app_id, isactive=True ).first() if not installed: return error_response(ErrorCodes.NOT_FOUND, 'Application not installed on this computer', http_code=404) installed.isactive = False db.session.commit() return success_response(message='Application uninstalled') @applications_bp.route('/machines//', methods=['PUT']) @require_permission('applications.edit') def update_installed_app(machine_id: int, app_id: int): """Update installed application (e.g., change version).""" models, err = _require_computer_models() if err: return err _, ComputerInstalledApp = models installed = ComputerInstalledApp.query.filter_by( computerid=machine_id, appid=app_id, isactive=True ).first() if not installed: return error_response(ErrorCodes.NOT_FOUND, 'Application not installed on this computer', http_code=404) data = request.get_json() if not data: return error_response(ErrorCodes.VALIDATION_ERROR, 'No data provided') if 'appversionid' in data: installed.appversionid = data['appversionid'] db.session.commit() return success_response(installed.to_dict(), message='Installation updated') # Support teams + contacts now live in the supportteams blueprint # (/api/supportteams), replacing the legacy appowners pair. # ============================================================================= # Image and installer uploads # # Both follow the model-image trio (upload / serve / delete). The image is # public because application pages render it before anything else loads; the # installer is not, because it is a binary a site pays for. # ============================================================================= @applications_bp.route('//image', methods=['POST']) @require_permission('applications.edit') def upload_application_image(app_id: int): """Upload (or replace) an application's image. multipart/form-data: file=. Stored as application-, one per application, and application.image is pointed at the served URL. """ app = db.session.get(Application, app_id) if not app: return error_response(ErrorCodes.NOT_FOUND, f'Application with ID {app_id} not found', http_code=404) 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 APP_IMAGE_EXTENSIONS: return error_response(ErrorCodes.VALIDATION_ERROR, f'Unsupported image type {ext}') filename = _replace_upload(_appimage_dir(), f'application-{app_id}', ext, upload) app.image = f'{APP_IMAGE_URL_PREFIX}{filename}' db.session.commit() AuditLog.log('updated', 'Application', entityid=app_id, entityname=app.appname, changes={'image': {'new': app.image}}) db.session.commit() return success_response(app.to_dict(), message='Application image uploaded') @applications_bp.route('/image/', methods=['GET']) def serve_application_image(filename): """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 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) @applications_bp.route('//image', methods=['DELETE']) @require_permission('applications.edit') def delete_application_image(app_id: int): """Remove an application's image and clear the field.""" app = db.session.get(Application, app_id) if not app: return error_response(ErrorCodes.NOT_FOUND, f'Application with ID {app_id} not found', http_code=404) for old in glob.glob(os.path.join(_appimage_dir(), secure_filename(f'application-{app_id}') + '.*')): os.remove(old) app.image = None db.session.commit() return success_response(app.to_dict(), message='Application image removed') @applications_bp.route('//package', methods=['POST']) @require_permission('applications.edit') def upload_application_package(app_id: int): """Upload (or replace) the installer for an application. multipart/form-data: file=. The original filename is kept in installpath so the download arrives named the way the vendor shipped it, and the stored copy is application-. """ app = db.session.get(Application, app_id) if not app: return error_response(ErrorCodes.NOT_FOUND, f'Application with ID {app_id} not found', http_code=404) 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 APP_PACKAGE_EXTENSIONS: return error_response( ErrorCodes.VALIDATION_ERROR, 'Unsupported installer type {}. Allowed: {}'.format( ext, ', '.join(sorted(APP_PACKAGE_EXTENSIONS)))) # Measure by seeking the stream, not Content-Length: a chunked upload has # no length header, and trusting the header lets a client understate size. upload.stream.seek(0, os.SEEK_END) size = upload.stream.tell() upload.stream.seek(0) if size > MAX_PACKAGE_BYTES: return error_response( ErrorCodes.VALIDATION_ERROR, 'Installer is {:.0f}MB; the limit is {:.0f}MB. Put larger payloads ' 'on the share and link to them with Install Path.'.format( size / 1048576, MAX_PACKAGE_BYTES / 1048576)) filename = _replace_upload(_apppackage_dir(), f'application-{app_id}', ext, upload) app.installpath = '{}{}'.format(APP_PACKAGE_URL_PREFIX, filename) # The endpoint installer push pins bytes by digest; compute it once here so # the server, not the agent, decides what the correct payload is. Stamp it # on the app itself so an update_app push can resolve an install source by # app alone (no AppVersion row). If the caller also names a version, stamp # it there too; otherwise return it so the version-edit flow can persist it. sha256 = _sha256_of(upload.stream) app.installersha256 = sha256 stamped_version = None appversionid = request.form.get('appversionid', type=int) if appversionid: stamped_version = db.session.get(AppVersion, appversionid) if not stamped_version or stamped_version.appid != app_id: return error_response( ErrorCodes.VALIDATION_ERROR, 'appversionid {} does not belong to this application'.format(appversionid)) stamped_version.sha256 = sha256 db.session.commit() AuditLog.log('updated', 'Application', entityid=app_id, entityname=app.appname, changes={'installpath': {'new': app.installpath}, 'uploadedfilename': {'new': upload.filename}, 'sha256': {'new': sha256}, 'bytes': {'new': size}}) db.session.commit() return success_response( {**app.to_dict(), 'uploadedfilename': upload.filename, 'bytes': size, 'sha256': sha256, 'appversionid': stamped_version.appversionid if stamped_version else None}, message='Installer uploaded') @applications_bp.route('/package/', methods=['GET']) @require_permission('applications.download') def serve_application_package(filename): """Download an uploaded installer. Authenticated, unlike the image: this is licensed vendor software, and an open URL would publish it to anything that can reach the site. Always sent as an attachment so a browser saves it rather than trying to render it. Gated on applications.download, not applications.view: a guest may be granted the metadata permission, and "an open URL would publish it" is just as true of a URL any anonymous visitor can reach. """ return send_upload(_apppackage_dir(), filename, as_attachment=True) @applications_bp.route('//package', methods=['DELETE']) @require_permission('applications.edit') def delete_application_package(app_id: int): """Remove an uploaded installer and clear the path it set.""" app = db.session.get(Application, app_id) if not app: return error_response(ErrorCodes.NOT_FOUND, f'Application with ID {app_id} not found', http_code=404) for old in glob.glob(os.path.join(_apppackage_dir(), secure_filename(f'application-{app_id}') + '.*')): os.remove(old) # Only clear the path if it pointed at the upload; a site's own share path # was typed by a person and is not ours to wipe. if (app.installpath or '').startswith(APP_PACKAGE_URL_PREFIX): app.installpath = None db.session.commit() return success_response(app.to_dict(), message='Installer removed')