"""Script library API endpoints. The ShopDB-hosted script library backs the endpoint run_script delivery path: a PowerShell script is authored/uploaded once here, and the endpoint agent fetches its body by digest on push (the same model as application installer delivery). The body never crosses the command wire and is never pre-staged on the endpoint; the agent downloads it with its collector credential and verifies sha256. Gated on the applications.* permissions - scripts are software delivery like apps, so they ride the same operator grant rather than a separate permission. """ import os from flask import Blueprint, request, current_app from shopdb.extensions import db from shopdb.core.models import Script, RUNCONTEXTS, RUNCONTEXT_SYSTEM, AuditLog from shopdb.utils.responses import ( success_response, error_response, ErrorCodes, ) from shopdb.utils.authz import require_permission from shopdb.utils.uploads import send_upload from shopdb.core.api.applications import _sha256_of, _replace_upload # Scripts are PowerShell for the endpoint agent. The guard is a sanity check on # the upload, not a security boundary - the download route always sends as an # attachment and the agent runs it explicitly. SCRIPT_EXTENSIONS = {'.ps1'} SCRIPT_URL_PREFIX = '/api/scripts/body/' # A delivery script is small; cap well under the installer limit. MAX_SCRIPT_BYTES = 1 * 1024 * 1024 def _script_dir(): return os.path.join(current_app.instance_path, 'scripts') scripts_bp = Blueprint('scripts', __name__) @scripts_bp.route('', methods=['GET']) @require_permission('applications.view') def list_scripts(): """List scripts (active by default).""" query = Script.query if request.args.get('active', 'true').lower() != 'false': query = query.filter(Script.isactive == True) # noqa: E712 if search := request.args.get('search'): query = query.filter(Script.scriptname.ilike(f'%{search}%')) rows = query.order_by(Script.scriptname).all() return success_response([row.to_dict() for row in rows]) @scripts_bp.route('/', methods=['GET']) @require_permission('applications.view') def get_script(script_id: int): script = db.session.get(Script, script_id) if not script: return error_response(ErrorCodes.NOT_FOUND, 'Script not found', http_code=404) return success_response(script.to_dict()) def _validate_runcontext(value): """Normalize/validate a runcontext, defaulting to system when absent.""" if value is None: return RUNCONTEXT_SYSTEM, None value = str(value).strip().lower() if value not in RUNCONTEXTS: return None, 'runcontext must be one of: {}'.format(', '.join(sorted(RUNCONTEXTS))) return value, None @scripts_bp.route('', methods=['POST']) @require_permission('applications.create') def create_script(): """Create a script catalog entry (body uploaded separately).""" data = request.get_json(silent=True) or {} name = (data.get('scriptname') or '').strip() if not name: return error_response(ErrorCodes.VALIDATION_ERROR, 'scriptname is required') if Script.query.filter_by(scriptname=name).first(): return error_response(ErrorCodes.VALIDATION_ERROR, f'A script named {name!r} already exists') runcontext, err = _validate_runcontext(data.get('runcontext')) if err: return error_response(ErrorCodes.VALIDATION_ERROR, err) script = Script( scriptname=name, scriptdescription=data.get('scriptdescription'), runcontext=runcontext, ) db.session.add(script) db.session.commit() AuditLog.log('created', 'Script', entityid=script.scriptid, entityname=name, changes={'runcontext': {'new': runcontext}}) db.session.commit() return success_response(script.to_dict(), message='Script created', http_code=201) @scripts_bp.route('/', methods=['PUT']) @require_permission('applications.edit') def update_script(script_id: int): script = db.session.get(Script, script_id) if not script: return error_response(ErrorCodes.NOT_FOUND, 'Script not found', http_code=404) data = request.get_json(silent=True) or {} if 'scriptname' in data: name = (data.get('scriptname') or '').strip() if not name: return error_response(ErrorCodes.VALIDATION_ERROR, 'scriptname cannot be empty') clash = Script.query.filter(Script.scriptname == name, Script.scriptid != script_id).first() if clash: return error_response(ErrorCodes.VALIDATION_ERROR, f'A script named {name!r} already exists') script.scriptname = name if 'scriptdescription' in data: script.scriptdescription = data.get('scriptdescription') if 'runcontext' in data: runcontext, err = _validate_runcontext(data.get('runcontext')) if err: return error_response(ErrorCodes.VALIDATION_ERROR, err) script.runcontext = runcontext if 'isactive' in data: script.isactive = bool(data.get('isactive')) db.session.commit() return success_response(script.to_dict(), message='Script updated') @scripts_bp.route('/', methods=['DELETE']) @require_permission('applications.delete') def delete_script(script_id: int): script = db.session.get(Script, script_id) if not script: return error_response(ErrorCodes.NOT_FOUND, 'Script not found', http_code=404) name = script.scriptname db.session.delete(script) db.session.commit() AuditLog.log('deleted', 'Script', entityid=script_id, entityname=name) db.session.commit() return success_response(None, message='Script deleted') @scripts_bp.route('//body', methods=['POST']) @require_permission('applications.edit') def upload_script_body(script_id: int): """Upload (or replace) the .ps1 body for a script. multipart/form-data: file=. Stored as script-.ps1 and the digest is stamped so the endpoint delivery path can pin exact bytes. """ script = db.session.get(Script, script_id) if not script: return error_response(ErrorCodes.NOT_FOUND, 'Script 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 SCRIPT_EXTENSIONS: return error_response( ErrorCodes.VALIDATION_ERROR, 'Unsupported script type {}. Allowed: {}'.format( ext, ', '.join(sorted(SCRIPT_EXTENSIONS)))) upload.stream.seek(0, os.SEEK_END) size = upload.stream.tell() upload.stream.seek(0) if size > MAX_SCRIPT_BYTES: return error_response( ErrorCodes.VALIDATION_ERROR, 'Script is {:.0f}KB; the limit is {:.0f}KB.'.format( size / 1024, MAX_SCRIPT_BYTES / 1024)) filename = _replace_upload(_script_dir(), f'script-{script_id}', ext, upload) script.filename = filename script.sha256 = _sha256_of(upload.stream) db.session.commit() AuditLog.log('updated', 'Script', entityid=script_id, entityname=script.scriptname, changes={'filename': {'new': filename}, 'sha256': {'new': script.sha256}, 'bytes': {'new': size}}) db.session.commit() return success_response( {**script.to_dict(), 'uploadedfilename': upload.filename, 'bytes': size}, message='Script body uploaded') @scripts_bp.route('/body/', methods=['GET']) @require_permission('applications.download') def download_script_body(filename: str): """Operator-facing script body download (JWT/RBAC). The endpoint agent uses the collector-authenticated plugin route instead; this one is for humans reviewing what a script does. Gated on applications.download, not applications.view: this serves the SYSTEM-context PowerShell a site pushes to its bays, and a guest may be granted the metadata permission. """ return send_upload(_script_dir(), filename, as_attachment=True)