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
import re

from flask import Blueprint, request, current_app

from shopdb.extensions import db
from shopdb.core.models import (Script, RUNCONTEXTS, RUNCONTEXT_SYSTEM,
                                ERRORMODES, ERRORMODE_DEFAULT, 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


def _validate_errormode(value):
    """Normalize/validate an errormode, defaulting to permissive when absent.

    Defaults to ERRORMODE_DEFAULT rather than strict on purpose. Strict promotes
    PowerShell's non-terminating errors to terminating ones, so a body that was
    written to tolerate a failing cleanup or to probe with a cmdlet that errors
    when the target is absent would begin aborting partway through. An absent
    field therefore means "what this script already did", never a silent upgrade.
    """
    if value is None:
        return ERRORMODE_DEFAULT, None
    value = str(value).strip().lower()
    if value not in ERRORMODES:
        return None, 'errormode must be one of: {}'.format(', '.join(sorted(ERRORMODES)))
    return value, None


def _validate_workingdirectory(value):
    """Normalize the optional folder a script runs in. Returns (value, error).

    Empty means "the agent's staging directory", which is the behaviour every
    script had before this field existed. The purpose is to let a script sit
    beside a corporate tool already installed on the endpoint and call it by
    relative path.

    The checks are shape checks, NOT a privilege boundary - a script body is
    already arbitrary SYSTEM PowerShell and can change its own directory. They
    exist so a typo fails here with an explanation instead of on the endpoint:

    - must be an absolute local path (a drive letter), because a relative one
      is meaningless without knowing what it would be relative TO
    - no UNC. The stated use case is a tool already on this machine; a UNC
      working directory resolves as the MACHINE ACCOUNT against a file server,
      which moves where the binaries come from and is a different trust story
      than "Intune put this here"
    - no traversal segment, which only ever obscures what the path really is
    """
    if value is None:
        return None, None
    value = str(value).strip().strip('"')
    if not value:
        return None, None
    if value.startswith('\\\\') or value.startswith('//'):
        return None, ('workingdirectory must be a local path on the endpoint, '
                      'not a UNC share')
    normalized = value.replace('/', '\\')
    if not re.match(r'^[A-Za-z]:\\', normalized):
        return None, ('workingdirectory must be an absolute path including a '
                      r'drive letter, for example C:\Program Files\Sysinternals')
    if '..' in normalized.split('\\'):
        return None, 'workingdirectory must not contain a ".." segment'
    if len(normalized) > 255:
        return None, 'workingdirectory must be 255 characters or fewer'
    # Trailing separator is noise, EXCEPT at a drive root: "C:\" trimmed to
    # "C:" is drive-RELATIVE on Windows and would resolve to wherever the
    # process happened to be on that drive, which is the opposite of pinning a
    # folder. Keep the root form intact.
    trimmed = normalized.rstrip('\\')
    if len(trimmed) < 3:
        trimmed = normalized[:3]
    return trimmed, 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)
    workingdirectory, err = _validate_workingdirectory(data.get('workingdirectory'))
    if err:
        return error_response(ErrorCodes.VALIDATION_ERROR, err)
    errormode, err = _validate_errormode(data.get('errormode'))
    if err:
        return error_response(ErrorCodes.VALIDATION_ERROR, err)

    script = Script(
        scriptname=name,
        scriptdescription=data.get('scriptdescription'),
        runcontext=runcontext,
        workingdirectory=workingdirectory,
        errormode=errormode,
    )
    db.session.add(script)
    db.session.commit()
    AuditLog.log('created', 'Script', entityid=script.scriptid,
                 entityname=name,
                 changes={'runcontext': {'new': runcontext},
                          'workingdirectory': {'new': workingdirectory},
                          'errormode': {'new': errormode}})
    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 'workingdirectory' in data:
        workingdirectory, err = _validate_workingdirectory(
            data.get('workingdirectory'))
        if err:
            return error_response(ErrorCodes.VALIDATION_ERROR, err)
        script.workingdirectory = workingdirectory
    if 'errormode' in data:
        errormode, err = _validate_errormode(data.get('errormode'))
        if err:
            return error_response(ErrorCodes.VALIDATION_ERROR, err)
        script.errormode = errormode
    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)