shopdb-flask/shopdb/core/api/scripts.py

"""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('/<int:script_id>', 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('/<int:script_id>', 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('/<int:script_id>', 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('/<int:script_id>/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=<script.ps1>. Stored as script-<id>.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/<path:filename>', 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)