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 endpoints.runscript, the same standalone grant that authorizes pushing
a script. These are NOT applications: an application installer is a vendor
artefact, while a script here is arbitrary PowerShell this site wrote that the
agent runs as SYSTEM. Gating library management on applications.* meant anyone
with the software-delivery grant could rewrite the bytes of a script a trusted
operator later pushes, which is a privilege escalation wearing a content-
management hat. The write and the push now require the same trust.

This TIGHTENS access, and there is deliberately NO backfill migration. The
operator scope grants applications.create/edit (they are edit verbs), so
operator-scope holders could reach this library before and cannot now. Do not
"fix" that by granting endpoints.runscript to applications.edit holders on
upgrade: it is in SCOPE_NEVER_GRANT precisely because it is a per-user checkbox
decision, and the same grant also authorizes the remote PUSH, so a backfill
would hand every operator SYSTEM code execution on the fleet. Admins are
unaffected via the role bypass; anyone else who genuinely needs the library
gets the checkbox in the user editor.
"""

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
# Owned by the endpoints plugin, which is where the risk is described in full
# ("ARBITRARY CODE, executes as SYSTEM"). Core names it as a string rather than
# importing it so this module stays importable with the plugin absent - the
# grant simply cannot be held by anyone if nothing seeds it.
SCRIPT_PERMISSION = 'endpoints.runscript'


def _script_dir():
    return os.path.join(current_app.instance_path, 'scripts')


scripts_bp = Blueprint('scripts', __name__)


@scripts_bp.route('', methods=['GET'])
@require_permission(SCRIPT_PERMISSION)
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(SCRIPT_PERMISSION)
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(SCRIPT_PERMISSION)
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(SCRIPT_PERMISSION)
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(SCRIPT_PERMISSION)
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(SCRIPT_PERMISSION)
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(SCRIPT_PERMISSION)
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.

    Same grant as the rest of the library: this serves the SYSTEM-context
    PowerShell a site pushes to its bays, so reading it is not a lesser act
    than listing it.
    """
    return send_upload(_script_dir(), filename, as_attachment=True)