0134eaa219
Soft delete left orphaned executables on disk and let a deleted name collide with a fresh import, so delete is now a hard delete that removes the associated files. The destructive paths (delete, revert-to-record) require a typed OK because neither can be undone. Installers keep their filename at import and nest per application/version, which stops repeat imports doubling up. Version now comes from the same PE detection the endpoints agent uses; when no version can be read the caller is asked for one rather than silently storing a blank. Applications also surface Version and a monitorable toggle, with launch path and process name required only while monitorable is ticked. Fleet shows the PC alias, matching PC Details.
1102 lines
42 KiB
Python
1102 lines
42 KiB
Python
"""Applications API endpoints."""
|
|
|
|
import glob
|
|
import hashlib
|
|
import re
|
|
import os
|
|
import shutil
|
|
|
|
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/'
|
|
|
|
|
|
def _requires_policy_ack(app, data):
|
|
"""True when this save would newly generate an install-enforcing policy.
|
|
|
|
Ticking "Required on all PCs" does not just flag a report row: it creates a
|
|
policy that installs this application on every matched endpoint. That is a
|
|
real change to real machines, so the caller has to acknowledge it
|
|
explicitly. Enforced here rather than only in the form because the checkbox
|
|
is reachable through the API too, and a confirmation dialog that can be
|
|
skipped by calling the endpoint directly is not a safeguard.
|
|
|
|
Only NEWLY ticking it needs acknowledgement. Unticking, or saving an app
|
|
that was already required, does not.
|
|
"""
|
|
if not data.get('isrequired'):
|
|
return False
|
|
if app is not None and app.isrequired:
|
|
return False
|
|
return not data.get('acknowledgepolicy')
|
|
|
|
|
|
POLICY_ACK_MESSAGE = (
|
|
'Marking this application as required on all PCs creates a configuration '
|
|
'policy that will install it on every non-server endpoint it matches. '
|
|
'Confirm to create the policy.'
|
|
)
|
|
|
|
# Deleting an application is irreversible and takes its versions and uploaded
|
|
# installer with it, so the caller has to type this rather than click once.
|
|
# Checked server-side as well as in the dialog: the endpoint is reachable
|
|
# directly, and a confirmation that only exists in the UI is not a safeguard.
|
|
DELETE_CONFIRM_TOKEN = 'OK'
|
|
|
|
DELETE_CONFIRM_MESSAGE = (
|
|
'Deleting an application permanently removes it, its versions, and its '
|
|
'uploaded installer. This cannot be undone. Type OK to confirm.'
|
|
)
|
|
|
|
|
|
def _delete_confirmed(data):
|
|
"""True when the caller typed the confirmation token exactly.
|
|
|
|
Compared case-sensitively after stripping surrounding whitespace. Accepting
|
|
"ok" would defeat the point of asking for deliberate input.
|
|
"""
|
|
if not isinstance(data, dict):
|
|
return False
|
|
return str(data.get('confirm', '')).strip() == DELETE_CONFIRM_TOKEN
|
|
|
|
|
|
def _validate_monitorable(data, existing=None):
|
|
"""Return an error message if monitorable is set without its two fields.
|
|
|
|
A process monitor needs an image name to watch for and a path to launch when
|
|
it is missing; without both, ticking the box would create a monitor that
|
|
cannot do anything. Enforced here rather than as a DB constraint so the
|
|
columns can stay nullable for rows that predate the feature.
|
|
|
|
`existing` lets a PATCH-style update validate the MERGED result: a payload
|
|
that only toggles the checkbox is still checked against values already
|
|
stored on the row.
|
|
"""
|
|
def merged(field):
|
|
if field in data:
|
|
return (data.get(field) or '').strip()
|
|
return ((getattr(existing, field, None) or '').strip()
|
|
if existing is not None else '')
|
|
|
|
if 'ismonitorable' in data:
|
|
monitorable = data.get('ismonitorable')
|
|
else:
|
|
monitorable = getattr(existing, 'ismonitorable', False)
|
|
|
|
if not monitorable:
|
|
return None
|
|
|
|
if not merged('monitorlaunchpath'):
|
|
return 'A monitorable application needs a launch path'
|
|
if not merged('monitorprocessname'):
|
|
return 'A monitorable application needs a process name'
|
|
return None
|
|
|
|
|
|
def _sync_required_app_policy(app):
|
|
"""Sync the auto-generated "required on all PCs" policy, if available.
|
|
|
|
The policy engine lives in the computers PLUGIN, which may be absent or
|
|
disabled, so this is a guarded import: core must not hard-depend on a
|
|
plugin. When the plugin is not installed the checkbox keeps its original
|
|
compliance-report meaning and nothing else happens.
|
|
"""
|
|
try:
|
|
from plugins.computers.services.apppolicy import sync_required_app_policy
|
|
except ImportError:
|
|
return None
|
|
try:
|
|
return sync_required_app_policy(app)
|
|
except Exception:
|
|
# A policy-generation failure must not lose the user's application edit.
|
|
current_app.logger.exception(
|
|
'Failed to sync the required-app policy for appid=%s', app.appid)
|
|
return None
|
|
|
|
# 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
|
|
|
|
# How much of an installer to read when looking for its version resource. The
|
|
# PE resource directory is usually within the first few MB; reading the whole
|
|
# file would mean holding a 500MB upload in memory to learn a version string.
|
|
PE_VERSION_SCAN_BYTES = 32 * 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 _detect_installer_version(stream):
|
|
"""Read the FileVersion out of an uploaded PE installer, or None.
|
|
|
|
Deliberately reuses the endpoints plugin's PE reader rather than parsing the
|
|
format again here. A version detected at upload has to match what the agent
|
|
reports from the installed machine, and two separate implementations of the
|
|
same structure walk would drift apart and make an endpoint look
|
|
out-of-compliance for no real reason.
|
|
|
|
FileVersion only, never ProductVersion: ProductVersion may carry semver
|
|
metadata (1.5.0+abc123) which is not a version number we can compare or use
|
|
in a path. Returns None for MSI and archives, which are not PE files at all
|
|
and simply have no version to read.
|
|
"""
|
|
try:
|
|
from plugins.endpoints.api.agentstore import read_pe_file_version
|
|
except ImportError:
|
|
return None
|
|
|
|
position = stream.tell()
|
|
try:
|
|
# The version resource lives in the PE header near the start, but the
|
|
# resource directory it points into can sit much deeper. Read a bounded
|
|
# prefix so a 500MB installer is not pulled into memory whole.
|
|
data = stream.read(PE_VERSION_SCAN_BYTES)
|
|
finally:
|
|
stream.seek(position)
|
|
|
|
# The parser documents itself as never raising, but it walks offsets read
|
|
# out of an attacker-supplied file. A version is a convenience; it must not
|
|
# cost the operator the upload of a 500MB installer if that guarantee ever
|
|
# slips.
|
|
try:
|
|
return read_pe_file_version(data)
|
|
except Exception:
|
|
current_app.logger.warning('Version detection failed for an installer',
|
|
exc_info=True)
|
|
return None
|
|
|
|
|
|
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
|
|
)
|
|
|
|
monitor_error = _validate_monitorable(data)
|
|
if monitor_error:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, monitor_error)
|
|
|
|
if _requires_policy_ack(None, data):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, POLICY_ACK_MESSAGE)
|
|
|
|
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),
|
|
ismonitorable=data.get('ismonitorable', False),
|
|
monitorlaunchpath=data.get('monitorlaunchpath'),
|
|
monitorprocessname=data.get('monitorprocessname'),
|
|
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)
|
|
|
|
policyresult = _sync_required_app_policy(app)
|
|
|
|
db.session.commit()
|
|
|
|
payload = app.to_dict()
|
|
if policyresult in ('created', 'removed', 'kept'):
|
|
payload['policyresult'] = policyresult
|
|
|
|
return success_response(payload, 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
|
|
)
|
|
|
|
monitor_error = _validate_monitorable(data, existing=app)
|
|
if monitor_error:
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, monitor_error)
|
|
|
|
if _requires_policy_ack(app, data):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, POLICY_ACK_MESSAGE)
|
|
|
|
fields = [
|
|
'appname', 'appdescription', 'supportteamid', 'isinstallable',
|
|
'applicationnotes', 'installpath', 'applicationlink', 'documentationpath',
|
|
'ishidden', 'isprinter', 'islicenced', 'isrequired', 'image', 'isactive',
|
|
'ismonitorable', 'monitorlaunchpath', 'monitorprocessname'
|
|
]
|
|
|
|
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)
|
|
|
|
# Only touch policy when the checkbox itself moved. Syncing on every save
|
|
# would let an unrelated edit resurrect a policy an admin had deleted.
|
|
policyresult = None
|
|
if 'isrequired' in changes:
|
|
policyresult = _sync_required_app_policy(app)
|
|
|
|
apply_import_timestamps(app, data)
|
|
db.session.commit()
|
|
|
|
payload = app.to_dict()
|
|
if policyresult in ('created', 'removed', 'kept'):
|
|
payload['policyresult'] = policyresult
|
|
|
|
return success_response(payload, message='Application updated')
|
|
|
|
|
|
def _remove_package_files(app_id: int):
|
|
"""Delete the stored installer(s) for an application, leaving the image.
|
|
|
|
Split out from _remove_upload_files because revert-to-record removes the
|
|
executables but keeps the icon: the record stays in the catalogue and a
|
|
record with no image is a worse record. Delete calls both.
|
|
|
|
Best-effort by design: a missing or unreadable file must not abort the
|
|
operation and strand the row. Returns the filenames actually removed so the
|
|
audit entry records what went.
|
|
"""
|
|
removed = []
|
|
packagedir = os.path.join(_apppackage_dir(), str(app_id))
|
|
for path in glob.glob(os.path.join(packagedir, '*')):
|
|
removed.append(os.path.basename(path))
|
|
if os.path.isdir(packagedir):
|
|
shutil.rmtree(packagedir, ignore_errors=True)
|
|
if os.path.isdir(packagedir):
|
|
current_app.logger.warning(
|
|
'Could not remove %s for appid=%s', packagedir, app_id)
|
|
return removed
|
|
|
|
|
|
def _remove_upload_files(app_id: int):
|
|
"""Delete the stored installer and image for an application.
|
|
|
|
Used by delete, which takes everything. Revert calls
|
|
_remove_package_files() instead so the image survives.
|
|
"""
|
|
removed = []
|
|
# The image is still one flat file per application; the installer now lives
|
|
# in a per-application directory so it can keep its vendor filename.
|
|
pattern = os.path.join(_appimage_dir(),
|
|
secure_filename(f'application-{app_id}') + '.*')
|
|
for path in glob.glob(pattern):
|
|
try:
|
|
os.remove(path)
|
|
removed.append(os.path.basename(path))
|
|
except OSError:
|
|
current_app.logger.warning(
|
|
'Could not remove %s while deleting appid=%s', path, app_id)
|
|
|
|
return removed + _remove_package_files(app_id)
|
|
|
|
|
|
def _clear_application_references(app_id: int):
|
|
"""Detach rows that point at this application before it is deleted.
|
|
|
|
The FKs differ in intent, so they are not treated the same:
|
|
- appversions.appid is NOT NULL and the versions are meaningless without
|
|
the application, so they go.
|
|
- computerinstalledapps.appid is NOT NULL; these are inventory links, so
|
|
the link goes while the computer stays.
|
|
- knowledgebase.appid is nullable and the article is somebody's writing,
|
|
so it is unlinked rather than deleted.
|
|
"""
|
|
AppVersion.query.filter_by(appid=app_id).delete(synchronize_session=False)
|
|
|
|
models = _computer_models()
|
|
if models is not None:
|
|
_, ComputerInstalledApp = models
|
|
ComputerInstalledApp.query.filter_by(appid=app_id).delete(
|
|
synchronize_session=False)
|
|
|
|
try:
|
|
from plugins.knowledgebase.models import KnowledgeBase
|
|
except ImportError:
|
|
return
|
|
KnowledgeBase.query.filter_by(appid=app_id).update(
|
|
{'appid': None}, synchronize_session=False)
|
|
|
|
|
|
@applications_bp.route('/<int:app_id>', methods=['DELETE'])
|
|
@require_permission('applications.delete')
|
|
def delete_application(app_id: int):
|
|
"""Permanently delete an application, its versions, and its uploads.
|
|
|
|
This is a hard delete. The previous soft delete only set isactive=False,
|
|
which left the name reserved forever: creating an application with the same
|
|
name failed with "already exists" against a row no screen would show.
|
|
"""
|
|
app = db.session.get(Application, app_id)
|
|
|
|
if not app:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Application not found', http_code=404)
|
|
|
|
if not _delete_confirmed(request.get_json(silent=True)):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, DELETE_CONFIRM_MESSAGE)
|
|
|
|
# Withdraw the generated policy first. A policy that outlived its
|
|
# application would keep installing something that no longer exists.
|
|
appname = app.appname
|
|
app.isrequired = False
|
|
_sync_required_app_policy(app)
|
|
|
|
_clear_application_references(app_id)
|
|
removed = _remove_upload_files(app_id)
|
|
|
|
db.session.delete(app)
|
|
|
|
# The audit row is the only record left that this application existed, so
|
|
# it names the files that went with it.
|
|
AuditLog.log('deleted', 'Application', entityid=app_id, entityname=appname,
|
|
changes={'removedfiles': removed} if removed else None)
|
|
|
|
db.session.commit()
|
|
|
|
return success_response(message='Application deleted')
|
|
|
|
|
|
REVERT_CONFIRM_MESSAGE = (
|
|
'Remove all policy associations and uploaded executables? This action '
|
|
"cannot be reversed. Type OK to proceed"
|
|
)
|
|
|
|
|
|
@applications_bp.route('/<int:app_id>/revert', methods=['POST'])
|
|
@require_permission('applications.edit')
|
|
def revert_application_to_record(app_id: int):
|
|
"""Strip an application back to a catalogue record.
|
|
|
|
The middle ground between editing and deleting. Deleting loses the name,
|
|
the description and the knowledgebase links, which is too much when the
|
|
intent is only to stop deploying something and re-upload a clean installer.
|
|
This withdraws the generated policy, drops the version rows and removes the
|
|
uploaded files, while the application itself stays in the catalogue.
|
|
|
|
Inventory links are deliberately kept: computerinstalledapps records that a
|
|
machine really does have this software on it, and that stays true whether
|
|
or not we still ship an installer for it. Deleting those would make the
|
|
fleet look clean when it is not.
|
|
|
|
Gated on applications.edit rather than applications.delete: the record
|
|
survives, and a user trusted to change an application is trusted to stop
|
|
deploying it.
|
|
"""
|
|
app = db.session.get(Application, app_id)
|
|
if not app:
|
|
return error_response(ErrorCodes.NOT_FOUND, 'Application not found',
|
|
http_code=404)
|
|
|
|
if not _delete_confirmed(request.get_json(silent=True)):
|
|
return error_response(ErrorCodes.VALIDATION_ERROR, REVERT_CONFIRM_MESSAGE)
|
|
|
|
# Withdraw the generated policy first, for the same reason delete does: a
|
|
# policy left behind would keep pushing an installer we just removed.
|
|
app.isrequired = False
|
|
_sync_required_app_policy(app)
|
|
|
|
AppVersion.query.filter_by(appid=app_id).delete(synchronize_session=False)
|
|
|
|
# Executables only. The image is not an executable and the warning does not
|
|
# offer to remove it - the record survives this operation, and stripping its
|
|
# icon would degrade the thing we are explicitly keeping.
|
|
removed = _remove_package_files(app_id)
|
|
|
|
# Only clear an installpath we own. A UNC path typed by an operator points
|
|
# at their share, not our upload directory, and is not ours to wipe.
|
|
if (app.installpath or '').startswith(APP_PACKAGE_URL_PREFIX):
|
|
app.installpath = None
|
|
app.installersha256 = None
|
|
|
|
AuditLog.log('updated', 'Application', entityid=app_id,
|
|
entityname=app.appname,
|
|
changes={'revertedtorecord': {'new': True},
|
|
'removedfiles': removed})
|
|
|
|
db.session.commit()
|
|
|
|
return success_response(app.to_dict(), message='Application reverted to record')
|
|
|
|
|
|
# ---- 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')
|
|
|
|
# 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('/<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'])
|
|
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 <img src="/api/applications/image/application-3.png"> 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('/<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')
|
|
|
|
|
|
def _store_package(app_id: int, upload):
|
|
"""Save an installer under its own vendor filename, one per application.
|
|
|
|
Keeping the shipped name matters at import: "DeviceLoggerSetup.msi" is
|
|
recognisable as the thing already in the catalogue, whereas the old
|
|
application-<id>.msi told the operator nothing and made a re-import look
|
|
like new software, which is how duplicate applications and versions got
|
|
created in the first place.
|
|
|
|
Files are namespaced by application id so two applications shipping a
|
|
generic "setup.exe" cannot overwrite each other. Any previous installer for
|
|
this application is removed first, so replacing one does not accumulate.
|
|
|
|
Returns (storedname, originalname).
|
|
"""
|
|
directory = os.path.join(_apppackage_dir(), str(app_id))
|
|
if os.path.isdir(directory):
|
|
shutil.rmtree(directory, ignore_errors=True)
|
|
os.makedirs(directory, exist_ok=True)
|
|
|
|
original = os.path.basename(upload.filename or '')
|
|
# secure_filename can reduce a name to nothing (all-unicode, all-dots), and
|
|
# an empty name would save as a bare directory entry.
|
|
stored = secure_filename(original) or 'installer{}'.format(
|
|
os.path.splitext(original)[1].lower())
|
|
|
|
upload.save(os.path.join(directory, stored))
|
|
return stored, original
|
|
|
|
|
|
@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 installer is stored under the
|
|
filename the vendor shipped, inside a per-application directory.
|
|
"""
|
|
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))
|
|
|
|
# Read the version before saving: the stream is consumed by save().
|
|
detectedversion = _detect_installer_version(upload.stream)
|
|
|
|
filename, _original = _store_package(app_id, upload)
|
|
app.installpath = '{}{}/{}'.format(APP_PACKAGE_URL_PREFIX, app_id, 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,
|
|
# None means "no version resource" - the form must ask the operator
|
|
# rather than silently saving an installer with no version, which is
|
|
# how duplicate version rows appeared.
|
|
'detectedversion': detectedversion,
|
|
'appversionid': stamped_version.appversionid if stamped_version else None},
|
|
message='Installer uploaded')
|
|
|
|
|
|
@applications_bp.route('/package/<path:filename>', 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.
|
|
|
|
filename is "<appid>/<vendorname>" now that installers keep the name they
|
|
shipped under. send_from_directory refuses to escape the base directory, so
|
|
the appid segment cannot be walked out of with "..".
|
|
"""
|
|
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)
|
|
|
|
shutil.rmtree(os.path.join(_apppackage_dir(), str(app_id)),
|
|
ignore_errors=True)
|
|
# 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
|
|
app.installersha256 = None
|
|
db.session.commit()
|
|
return success_response(app.to_dict(), message='Installer removed')
|