Files
shopdb-flask/shopdb/core/api/applications.py

651 lines
23 KiB
Python

"""Applications API endpoints."""
import glob
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
# 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 _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('/<int:app_id>', 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('/<int:app_id>', 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('/<int:app_id>', 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('/<int:app_id>/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('/<int:app_id>/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')
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=data.get('releasedate'),
notes=data.get('notes')
)
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('/<int:app_id>/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/<int:machine_id>', 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/<int:machine_id>', 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/<int:machine_id>/<int:app_id>', 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/<int:machine_id>/<int:app_id>', 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('/<int:app_id>/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=<image>. Stored as application-<id><ext>, 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/<path:filename>', methods=['GET'])
@require_permission('applications.view')
def serve_application_image(filename):
"""Serve an uploaded application image (public - lists and tiles read it)."""
return send_upload(_appimage_dir(), filename)
@applications_bp.route('/<int:app_id>/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('/<int:app_id>/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=<installer>. The original filename is kept in
installpath so the download arrives named the way the vendor shipped it,
and the stored copy is application-<id><ext>.
"""
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)
db.session.commit()
AuditLog.log('updated', 'Application', entityid=app_id,
entityname=app.appname,
changes={'installpath': {'new': app.installpath},
'uploadedfilename': {'new': upload.filename},
'bytes': {'new': size}})
db.session.commit()
return success_response(
{**app.to_dict(), 'uploadedfilename': upload.filename, 'bytes': size},
message='Installer uploaded')
@applications_bp.route('/package/<path:filename>', methods=['GET'])
@require_permission('applications.view')
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.
"""
return send_upload(_apppackage_dir(), filename, as_attachment=True)
@applications_bp.route('/<int:app_id>/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')