shopdb-flask/plugins/endpoints/commands.py

"""Server-side allowlist of endpoint agent commands.

The WebUI never sends an arbitrary command string. It sends an allowlisted
ACTION NAME; the concrete on-endpoint behavior (which script to run, how to stop
a process) lives in the agent. This module is the single source of truth for
which action names are legal and what they mean, so both the enqueue route and
the settings/UI surface validate against the same set.

The structure anticipates per-PC-type allowlists: today every computer may run
every action in COMMAND_ALLOWLIST, but ALLOWED_ACTIONS_FOR is the seam a later
change swaps for a type-scoped lookup (inspection PCs vs shopfloor PCs).
"""

# action name -> operator-facing metadata. label/description feed the UI; the
# agent maps the action name to its local handler (see CommandExecutor).
#
# Two families of action:
#   * Named MTConnect actions - fixed target, no parameters. The original,
#     bay-specific controls kept for the inline services-panel toggle.
#   * Generic RMM actions (service_* / process_*) - the target is supplied by
#     the policy action's params ('target' = a Windows service name or a process
#     image like 'DncMain.exe'). This is what populates the Configuration Policy
#     builder with reusable controls that any scope (PC type or application) can
#     attach. The 'targetkind' hint tells the builder which target field to show.
#     NOTE: agent-side execution of the generic actions is not wired yet; the
#     server allowlist and policy plumbing land first (the whole feature is
#     flag-gated), so a generic action saved today is authored and gated but is
#     a no-op on the endpoint until the CommandExecutor handlers ship.
COMMAND_ALLOWLIST = {
    'service_start': {
        'label': 'Start service',
        'description': 'Start a Windows service by name on the endpoint.',
        'targetkind': 'service',
    },
    'service_stop': {
        'label': 'Stop service',
        'description': 'Stop a running Windows service by name on the endpoint.',
        'targetkind': 'service',
    },
    'service_status': {
        'label': 'Check service',
        'description': 'Report whether a named Windows service is running.',
        'targetkind': 'service',
    },
    'process_start': {
        'label': 'Start process',
        'description': 'Launch a process by executable path on the endpoint.',
        'targetkind': 'process',
    },
    'process_stop': {
        'label': 'Stop process',
        'description': 'Stop a running process by image name on the endpoint.',
        'targetkind': 'process',
    },
    'process_status': {
        'label': 'Check process',
        'description': 'Report whether a named process is running.',
        'targetkind': 'process',
    },
    'mtconnect_start': {
        'label': 'Start MTConnect',
        'description': 'Run the MTConnect UDC agent start script on the endpoint.',
    },
    'mtconnect_stop': {
        'label': 'Stop MTConnect',
        'description': 'Stop the running MTConnect UDC agent on the endpoint.',
    },
    'mtconnect_status': {
        'label': 'Check MTConnect',
        'description': 'Report whether the MTConnect UDC agent is running.',
    },
    'system_restart': {
        'label': 'Restart computer',
        'description': 'Restart the endpoint computer immediately.',
    },
    'collect_now': {
        'label': 'Collect now',
        'description': 'Force the endpoint agent to report fresh telemetry off-cadence.',
    },
    'network_connections': {
        'label': 'Show network connections',
        'description': 'Return the full TCP connection table with owning processes. '
                       'Read-only diagnostic; the standard report carries only '
                       'listening ports because established connections churn.',
    },
    'scheduledtask_run': {
        'label': 'Run scheduled task',
        'description': 'Start an already-registered Windows scheduled task on '
                       'the endpoint by its task path (for example '
                       '\\ReportIP). The agent does not create, modify, or '
                       'define tasks - it only starts one that already exists, '
                       'and only when the path clears the agent-side approved '
                       'task allowlist. The task runs in whatever context its '
                       'own registered definition specifies, not the agent\'s.',
        'targetkind': 'scheduledtask',
    },
    'run_script': {
        'label': 'Run script',
        'description': 'Run a ShopDB-hosted script on the endpoint. The agent '
                       'fetches the script body by digest using its collector '
                       'credential and runs it in the script\'s configured '
                       'context (system or the logged-on user). CAUTION: a '
                       'script sent this way runs as SYSTEM unless its '
                       'descriptor pins the user context, so this is arbitrary '
                       'code execution on the endpoint by design. Grant '
                       'endpoints.runscript only to the most trusted '
                       'operators, '
                       'and gate the UI behind an explicit confirm the way '
                       'system_restart is gated (see confirmRestart in '
                       'PCDetail.vue).',
        'targetkind': 'script',
    },
    'collect_log': {
        'label': 'Collect log file',
        'description': 'Read a plaintext log file from a known path on the '
                       'endpoint and return its contents to ShopDB for '
                       'viewing and download. Read-only: the agent opens the '
                       'file for reading with sharing permitted (so a log a '
                       'running application holds open can still be read), '
                       'never writes, and never deletes. Only the TAIL of a '
                       'file is returned, and only for plaintext extensions - '
                       'a binary or oversized file is refused rather than '
                       'mangled. Gated on endpoints.updateapp because the '
                       'reason to collect a log is almost always to diagnose '
                       'an application push, so it belongs with app lifecycle '
                       'rights rather than as a separate grant. Note this '
                       'still reads a caller-supplied path as SYSTEM: it is '
                       'bounded to text files, but it is not a sandbox.',
        'targetkind': 'logpath',
        # Introduced in agent 1.16.0. An older agent has no case for this action
        # and falls through to its unknown-action default, which answers
        # "Unknown or disallowed action" - a message that reads as an
        # authorization refusal and sends the operator looking at permissions
        # they already hold. Declaring the floor here lets the enqueue route
        # refuse first, naming both versions, so the answer is the actionable
        # one: this endpoint needs a newer agent.
        'minagentversion': '1.16.0',
    },
    'update_app': {
        'label': 'Install/update application',
        'description': 'Push a ShopDB-hosted installer for an application (by app '
                       'or by a specific version). The agent fetches the pinned '
                       'package by digest using its collector credential and runs '
                       'it silently.',
        'targetkind': 'app',
    },
}


# Dell Command | Update: TWO utilities, deliberately two separate actions.
# A scan reports what is available and changes nothing; an update installs and
# can reboot. Collapsing them into one parameterized action would make the
# destructive one reachable by editing a field, so they stay distinct and the
# UI can gate them differently.
#
# Both are driven through scheduledtask_run against ShopDB-owned task
# definitions (\ShopDB-DellCommandScan and \ShopDB-DellCommandUpdate, already in
# ScheduledTaskPolicy's shipped defaults) rather than a bespoke verb. The task
# definition - not the agent - pins the dcu-cli.exe path, its arguments, and its
# output path, which is why no new agent code is needed to run them.
#
# OUTPUT PATH CONVENTION. A ShopDB-initiated scan MUST write its report under
# %ProgramData%\ShopDB\EndpointAgent\ (the same SYSTEM-owned root as the agent
# logs), NOT to Dell's default location. Corporate-managed DCU tasks write to
# the default; keeping ShopDB's output separate is the only thing that makes
# "did ShopDB ask for this scan, or did corporate" answerable when the results
# are collected. The path lives in the task definition and is generated
# server-side, so it can change without touching the agent.
#
# STILL TO DO - the results collector. Firing DCU is the easy half. schtasks
# /run returns as soon as the task STARTS, so a successful scheduledtask_run
# means "scan started", never "scan finished" - the command result cannot carry
# findings. Reporting available updates back into ShopDB needs a collector that
# reads the report from the path above and folds it into the telemetry payload.
# That is a separate change from this one; do not infer from a green command
# result that the data path exists.


# Reserved seam - one-click mapped-drive remediation. NOT implemented; this is a
# placement note so the next person adds it here rather than inventing a new
# transport. When it lands it will re-establish a Disconnected mapping surfaced
# by the user-context card:
#
#     'remap_drive': {
#         'label': 'Reconnect mapped drive',
#         'description': 'Re-establish a disconnected network drive mapping.',
#         'targetkind': 'drive',
#     },
#
# It MUST execute in the interactive USER session. A SYSTEM-context remap creates
# a mapping in SYSTEM's own drive table that the operator cannot see, so the card
# would keep reporting the drive as broken while the "fix" silently succeeded
# somewhere nobody is looking. An ELEVATED remap fails the same way, because UAC
# token separation gives the elevated context a different drive table - the same
# reason the probe itself runs unelevated from the tray.
#
# CommandExecutor.RunScriptInUserSession is the existing path for this.


# Actions that demand a permission stricter than the base endpoints.command.
# The enqueue route requires the caller to ALSO hold the mapped permission, so a
# destructive action (a remote reboot) is a heavier grant than a routine toggle.
#
# These four are gated by SCOPE, never by Configuration Policy. Policy shapes
# what a PC TYPE exposes; it is not the authority on whether a given person may
# reboot a machine or run arbitrary code. The check runs in the enqueue route
# BEFORE the policy gate, so it is also not break-glass-able: an admin may
# override what policy says about a machine, never what permissions say about
# the caller.
#
# Previously scheduledtask_run, run_script and update_app all mapped to a single
# endpoints.deploy grant. That bundle made app updates impossible to grant
# without also granting arbitrary SYSTEM code execution via run_script, so they
# are now four independent permissions. They are NOT equal risk, and the
# descriptions in plugin.py get_permissions() say so plainly:
#   scheduledtask_run - starts something already registered and allowlisted
#                       agent-side
#   update_app        - runs a pinned installer
#   run_script        - arbitrary code as SYSTEM unless the descriptor pins a
#                       user context
#
# collect_log rides endpoints.updateapp rather than taking a grant of its own.
# The reason to read a log is nearly always to find out why an app push behaved
# the way it did, so the right to install an application and the right to read
# that application's log are the same job. It is read-only and strictly weaker
# than the installer push the same permission already allows.
ELEVATED_ACTION_PERMISSIONS = {
    'system_restart': 'endpoints.restart',
    'scheduledtask_run': 'endpoints.corptools',
    'run_script': 'endpoints.runscript',
    'update_app': 'endpoints.updateapp',
    'collect_log': 'endpoints.updateapp',
}


def elevated_permission_for(action):
    """The extra permission an action requires, or None for routine actions."""
    return ELEVATED_ACTION_PERMISSIONS.get(action)


# Shipped scripts that endpoints.corptools may push without holding
# endpoints.runscript.
#
# This is an allowance on ONE NAMED SCRIPT, never on the run_script action. The
# distinction is the whole point: relaxing the action would let any corptools
# holder push ANY script in the library - including the agent self-upgrade, and
# anything an operator uploads later - because the script id rides in the
# caller-controlled request body. That would silently rebuild the endpoints.deploy
# bundle the permission split took apart, which is the one outcome this must not
# produce.
#
# Granting on a name is only defensible because these names are MANAGED: each is
# seeded from a source-controlled template, pinned by SHA-256 on delivery, and
# re-asserted on every seed (see reportipscript.py). The name therefore cannot be
# made to mean something else by editing a row. An entry here is a statement that
# the reviewed body is operationally equivalent to something the user could
# already do unaided - Report IP is an icon on every Public Desktop in the
# company, so pushing it grants no capability the person did not already have.
#
# Do NOT add a name here because a denial is inconvenient. The test for
# membership is that a non-privileged employee can already trigger the same work
# by hand on the machine in front of them. Anything else needs endpoints.runscript.
CORPTOOLS_PUSHABLE_SCRIPTS = frozenset({
    'Report IP - Desktop Shortcut',
})


def corptools_may_push_script(scriptname):
    """True when endpoints.corptools alone is enough to push this script.

    Matched on the exact managed name. Callers must resolve the target to a real
    Script row FIRST and pass the stored scriptname, never a caller-supplied
    string: matching what the requester typed would let anyone name their way
    past the grant.
    """
    return scriptname in CORPTOOLS_PUSHABLE_SCRIPTS


def minimum_agent_version_for(action):
    """The lowest agent version that implements an action, or None if any will do."""
    return COMMAND_ALLOWLIST.get(action, {}).get('minagentversion')


def action_minimum_versions():
    """Every action that has an agent version floor, as {action: version}.

    Actions without a floor are omitted rather than mapped to None: a caller
    asking "is there a minimum for this?" gets a plain absent key, and the
    payload stays small as the allowlist grows.
    """
    return {
        action: spec['minagentversion']
        for action, spec in COMMAND_ALLOWLIST.items()
        if spec.get('minagentversion')
    }


def parse_agent_version(value):
    """A reported agent version as a comparable tuple, or None if unusable.

    Agents report four parts ('1.15.0.0') while the csproj and the binary store
    carry three ('1.16.0'), so a plain string compare puts a NEWER agent below
    the floor it satisfies. Both are normalised to a tuple and compared
    numerically, with missing trailing parts treated as zero, so '1.16.0' and
    '1.16.0.0' are the same version.

    Anything non-numeric returns None rather than a guess. The fleet carries
    '0.0.0-sample' on seeded rows, and there is no honest ordering between a
    placeholder and a real build - the caller decides what to do about an
    unknown version, which is a different question from being out of date.
    """
    if not value:
        return None
    parts = []
    for chunk in str(value).strip().split('.'):
        if not chunk.isdigit():
            return None
        parts.append(int(chunk))
    if not parts:
        return None
    while len(parts) < 4:
        parts.append(0)
    return tuple(parts[:4])


def agent_meets_minimum(reported, minimum):
    """Whether a reported agent version satisfies a minimum.

    Returns False when either side is unparseable. That is the fail-closed
    direction: an endpoint whose version we cannot read is one we cannot claim
    supports the action, and letting it through only moves the failure to the
    agent where the error message is worse.
    """
    if not minimum:
        return True
    reportedparts = parse_agent_version(reported)
    minimumparts = parse_agent_version(minimum)
    if reportedparts is None or minimumparts is None:
        return False
    return reportedparts >= minimumparts


# Maps a monitored services/processes row to the start/stop/status actions that
# control it, so the services panel can render an inline toggle on exactly the
# rows that have a registered command action. The key is the agent-reported
# 'app' label; 'kind' pins the toggle to the authoritative row (for MTConnect
# the agent.exe PROCESS is the source of truth - run.bat launches it and the
# Windows service state is not reliable), so the service row stays badge-only.
ROW_COMMAND_TARGETS = {
    ('MTConnect', 'process'): {
        'start': 'mtconnect_start',
        'stop': 'mtconnect_stop',
        'status': 'mtconnect_status',
    },
}


def row_actions_for(app, kind, name=None):
    """Return the {start, stop, status, target, targetkind} map for a row.

    Two paths:
      * A row registered in ROW_COMMAND_TARGETS (MTConnect) keeps its fixed,
        named actions with no wire target - agent.exe is the source of truth and
        the behavior is pinned in the agent.
      * Any other process/service row maps to the generic primitives
        (process_* / service_*) with the row's OWN reported name as the wire
        target. This is what lets a new app light up start/stop/status without
        an agent rebuild: the target rides the wire.

    Rows with no usable kind return None and render as a plain status badge.
    """
    if not app and not name:
        return None
    fixed = ROW_COMMAND_TARGETS.get((app, (kind or '').lower()))
    if fixed:
        return {**fixed, 'target': None, 'targetkind': None}
    normalized = (kind or '').lower()
    if normalized == 'process' and name:
        return {
            'start': 'process_start',
            'stop': 'process_stop',
            'status': 'process_status',
            'target': name,
            'targetkind': 'process',
        }
    if normalized == 'service' and name:
        return {
            'start': 'service_start',
            'stop': 'service_stop',
            'status': 'service_status',
            'target': name,
            'targetkind': 'service',
        }
    return None


def is_allowed_action(action):
    """True when action is a known, allowlisted command name."""
    return action in COMMAND_ALLOWLIST


def action_targetkind(action):
    """The targetkind an action requires ('service'/'process'), or None.

    Generic actions declare a targetkind in COMMAND_ALLOWLIST; fixed-target
    actions (mtconnect_*) declare none and must not carry a wire target.
    """
    meta = COMMAND_ALLOWLIST.get(action) or {}
    return meta.get('targetkind')


def requires_target(action):
    """True when the action is generic and needs a wire target."""
    return action_targetkind(action) is not None


# Monitor attachment as a policy action. A monitor definition is bound to a PC
# type by carrying a ConfigurationPolicyAction whose name is 'monitor:<id>'.
# Distinct names let one policy attach several monitors and fold naturally
# (first-match-wins per name), and they are validated in the policy layer but
# never enqueued as commands because ALLOWED_ACTIONS_FOR intersects with
# COMMAND_ALLOWLIST, which carries no monitor: entries.
MONITOR_ACTION_PREFIX = 'monitor:'


def monitor_action_name(endpointmonitorid):
    """Build the policy action name that attaches a monitor definition."""
    return f'{MONITOR_ACTION_PREFIX}{endpointmonitorid}'


def is_monitor_action(action):
    """True when action names a monitor attachment (monitor:<id>)."""
    return isinstance(action, str) and action.startswith(MONITOR_ACTION_PREFIX)


def monitor_id_from_action(action):
    """The monitor id an action attaches, or None when it is not a monitor action."""
    if not is_monitor_action(action):
        return None
    raw = action[len(MONITOR_ACTION_PREFIX):]
    try:
        return int(raw)
    except (TypeError, ValueError):
        return None


# Actions that are NOT a per-endpoint configuration concern.
#
# Membership here is a statement about the action's nature, not its risk. An
# action belongs in this set when Configuration Policy has nothing meaningful to
# say about it: it invokes an already-registered, agent-allowlisted task rather
# than describing how a machine should be shaped. Policy exists to shape what a
# given PC TYPE exposes - which monitors run, which service primitives are
# reachable - so a policy that never mentions such an action must not be read as
# removing it.
#
# This is deliberately narrow. Do not add an action here merely because a denial
# is inconvenient; the correct remedy for an over-narrow policy is to fix the
# policy, or to use the audited admin break-glass override at the enqueue route.
#
# Without this exemption the intersection at the end of ALLOWED_ACTIONS_FOR
# silently drops these actions on any endpoint whose effective policy is
# non-empty, producing "Action not permitted by policy for this endpoint" even
# for a caller who holds the matching grant. Exemption from POLICY is not
# exemption from PERMISSION: each of these actions still carries its own entry
# in ELEVATED_ACTION_PERMISSIONS, enforced per-action in the enqueue route
# before policy is ever consulted.
POLICY_EXEMPT_ACTIONS = frozenset({
    'scheduledtask_run',
    # Read-only telemetry refresh. Policy shapes WHAT a PC type exposes; asking
    # an agent to report its existing state off-cadence changes nothing about
    # the machine, so a policy that never mentions it must not be read as
    # removing it. Without this, "Refresh Status" dies on every endpoint whose
    # policy is non-empty - a denial no operator can act on, because the fix
    # lives in a policy that was never about this action.
    'collect_now',
    # Read-only diagnostic retrieval. Policy describes how a machine should be
    # shaped - which monitors run, which service primitives are reachable - and
    # reading a log file shapes nothing. An existing policy was authored before
    # this action existed and so can never mention it, which under the
    # intersection means every endpoint carrying any policy at all refuses log
    # collection, including for an admin. That denial is also unactionable in
    # the way that matters: logs are what an operator reaches for when an
    # endpoint is already misbehaving, so gating them behind editing that same
    # endpoint's policy withholds the evidence exactly when it is needed.
    # Exemption from POLICY is not exemption from PERMISSION - collect_log
    # still requires endpoints.updateapp via ELEVATED_ACTION_PERMISSIONS, and
    # the agent still refuses UNC, traversal, directories and non-text files.
    'collect_log',
    # Script push is gated by WHO the caller is, not by what a PC type exposes.
    # Configuration Policy describes machine shape; it has no view on whether a
    # given person may execute code, so a policy that never mentions run_script
    # must not be read as withdrawing it. Leaving it policy-gated forced sites
    # to keep an all-PCs "allow run_script" policy row alive purely to avoid
    # losing the action everywhere - a rule that encoded no real intent and
    # whose deletion silently disarmed Script Push on every policied endpoint.
    # Authority lives on the caller axis instead: run_script maps to
    # endpoints.runscript in ELEVATED_ACTION_PERMISSIONS, checked with
    # caller_holds at the enqueue route BEFORE policy is consulted, so a scoped
    # personal token cannot inherit it from an admin owner. This exemption
    # widens WHERE the action may run, never WHO may run it.
    'run_script',
})


def ALLOWED_ACTIONS_FOR(pctype=None, assetid=None):
    """The action names an endpoint may run.

    Base allowlist is flat (every action for every computer). When the computers
    Configuration/Business Policy feature is enabled AND an assetid is given, the
    result is narrowed to the actions the endpoint's effective policy permits -
    this is the allowlist-gating half of the policy design. With the feature off
    (or no assetid), it returns the full base allowlist, so nothing changes until
    a site opts in. Actions in POLICY_EXEMPT_ACTIONS are never narrowed away,
    because policy is not the authority on them; they remain gated by their own
    entry in ELEVATED_ACTION_PERMISSIONS at the enqueue route. Policy resolution
    is imported lazily to keep this module free of a hard dependency on the
    computers plugin.

    This answers allowlist questions only. It does not consult desired-state
    holds, and it is not a permission check - callers must still enforce
    ELEVATED_ACTION_PERMISSIONS themselves.
    """
    base = set(COMMAND_ALLOWLIST.keys())
    if assetid is None:
        return base
    try:
        from plugins.computers.services.policyresolver import effective_policy_for
    except Exception:
        return base
    effective = effective_policy_for(assetid)
    allowed = effective.allowed_actions()
    if not allowed:
        # No policy spoke for this endpoint. With the feature off the resolver
        # returns empty and we fall back to the base allowlist (unchanged
        # behavior). This is intentionally permissive; deny-by-default would be
        # a separate, explicit setting.
        return base
    # Attaching a monitor expresses intent to watch a process/service AND to
    # control it: the panel renders that row's Start/Stop/Status buttons, so the
    # gate must permit the matching generic primitives or every click 409s
    # "not permitted by policy". Expand each monitor:<id> to its kind's control
    # actions before intersecting.
    allowed = allowed | _monitor_control_actions(allowed)
    # Exempt actions are added back unconditionally: policy narrows the
    # per-endpoint surface, but it is not the authority on these actions. Their
    # own elevated permission still gates them in the enqueue route.
    allowed = allowed | POLICY_EXEMPT_ACTIONS
    # Never widen past the code-level allowlist: policy can only gate within it.
    return base & allowed


def _monitor_control_actions(action_names):
    """Control primitives granted by the monitor: attachments in a policy.

    A process monitor grants process_start/stop/status; a service monitor grants
    service_start/stop/status. The monitor's kind is read from the catalog, so a
    definition's kind is the single source of truth for what its row can do.
    """
    ids = set()
    for name in action_names:
        if isinstance(name, str) and name.startswith(MONITOR_ACTION_PREFIX):
            monitor_id = monitor_id_from_action(name)
            if monitor_id is not None:
                ids.add(monitor_id)
    if not ids:
        return set()

    try:
        from .models import EndpointMonitor
    except Exception:
        return set()

    granted = set()
    for monitor in EndpointMonitor.query.filter(
            EndpointMonitor.endpointmonitorid.in_(ids)).all():
        kind = (monitor.kind or '').lower()
        if kind == 'process':
            granted |= {'process_start', 'process_stop', 'process_status'}
        elif kind == 'service':
            granted |= {'service_start', 'service_stop', 'service_status'}
    return granted


def PLACE_POLICY_HOLD(assetid, action, reason, userid=None, username=None,
                      minutes=None):
    """Suspend desired-state reconciliation of one action on one endpoint.

    Called after an admin break-glasses past a policy denial, or directly from
    the admin hold dialog. Without this, an override on a desired-mode action is
    undone by the next reconcile pass: the admin stops the service, the resolver
    sees drift, and the command comes straight back. Gate-mode overrides need no
    hold - nothing reverses them.

    minutes=None takes DEFAULT_HOLD_MINUTES. minutes=0 places an INDEFINITE hold
    (expiresat null) that never clears itself and stands until released by hand.
    Nothing surfaces a stale indefinite hold, so it suppresses reconciliation of
    that action silently and forever - deliberate, but the sharpest edge here.

    Idempotent on an action already actively held: re-holding returns True
    without stacking a second row. The multi-select picker makes "select all"
    natural, and duplicate rows would mean release had to be pressed once per
    hidden duplicate to actually resume policy.

    Returns True when a hold is in place for the action afterwards. Resolved
    lazily against the computers plugin, matching ALLOWED_ACTIONS_FOR, so this
    module keeps no hard dependency on it: with the computers plugin absent
    there is no reconciliation to suppress, and False is the honest answer.
    """
    try:
        from plugins.computers.models import (PolicyActionHold,
                                              DEFAULT_HOLD_MINUTES)
        from shopdb.api import db
    except Exception:
        return False

    from datetime import datetime, timedelta
    from sqlalchemy import or_

    existing = (PolicyActionHold.query
                .filter(PolicyActionHold.assetid == assetid,
                        PolicyActionHold.action == action,
                        PolicyActionHold.isactive.is_(True),
                        or_(PolicyActionHold.expiresat.is_(None),
                            PolicyActionHold.expiresat > datetime.utcnow()))
                .first())
    if existing is not None:
        return True

    if minutes is None:
        expiresat = datetime.utcnow() + timedelta(minutes=DEFAULT_HOLD_MINUTES)
    elif int(minutes) == 0:
        expiresat = None
    else:
        expiresat = datetime.utcnow() + timedelta(minutes=int(minutes))

    hold = PolicyActionHold(
        assetid=assetid,
        action=action,
        expiresat=expiresat,
        createdbyuserid=userid,
        createdbyusername=username,
        reason=reason,
    )
    db.session.add(hold)
    db.session.commit()
    return True


def ACTIVE_POLICY_HOLDS(assetid):
    """The unexpired break-glass holds on one endpoint, newest first.

    The read counterpart to PLACE_POLICY_HOLD, resolved lazily against the
    computers plugin for the same reason: with that plugin absent there is no
    reconciliation to suppress, so there are no holds and an empty list is the
    honest answer.

    Expiry is filtered here rather than by a sweeper, matching the resolver, so
    a stalled job can never silently extend a hold. Expired rows stay in the
    table as the audit record; they are simply not reported as active.
    """
    try:
        from plugins.computers.models import PolicyActionHold
    except Exception:
        return []

    from datetime import datetime
    from sqlalchemy import or_

    # Null expiresat is an indefinite hold and is always active, so it is
    # matched explicitly - a bare comparison would report it as gone while the
    # resolver still honours it, which is the worst kind of disagreement.
    rows = (PolicyActionHold.query
            .filter(PolicyActionHold.assetid == assetid,
                    PolicyActionHold.isactive.is_(True),
                    or_(PolicyActionHold.expiresat.is_(None),
                        PolicyActionHold.expiresat > datetime.utcnow()))
            .all())
    # Sorted here rather than in SQL: NULLS FIRST is not portable across the
    # dialects this runs on. Indefinite holds sort ahead of every dated one,
    # then latest expiry first. Keyed on datetime.max rather than a numeric
    # fallback: datetime.min.timestamp() raises on Windows, and the placeholder
    # is never actually compared against a real date anyway - the leading flag
    # already separates the two groups.
    rows.sort(key=lambda row: (row.expiresat is None,
                               row.expiresat or datetime.max),
              reverse=True)
    return [{
        'action': row.action,
        'expiresat': row.expiresat.isoformat() if row.expiresat else None,
        'reason': row.reason,
        'createdbyusername': row.createdbyusername,
    } for row in rows]


def HOLDABLE_ACTIONS_FOR(assetid):
    """Desired-mode actions on this endpoint, with their current hold state.

    Only desired-mode actions are offered: a hold suspends desired-state
    reconciliation, so holding a gate-mode action would suspend nothing and the
    control would silently do nothing.

    Returns [] when the computers plugin is absent or policies are off - with no
    reconciliation running there is nothing to hold, and an empty picker is the
    honest answer rather than a list of actions the hold would not affect.
    """
    try:
        from plugins.computers.services.policyresolver import effective_policy_for
    except Exception:
        return []

    effective = effective_policy_for(assetid)
    held = effective.heldactions
    return [{
        'action': name,
        'held': name in held,
    } for name in sorted(effective.holdable_actions())]


def RELEASE_POLICY_HOLD(assetid, action):
    """End an active hold early, resuming reconciliation of one action.

    Soft-deletes via isactive rather than dropping the row: the hold IS the
    audit record of what was suppressed and by whom, and deleting it on release
    would erase exactly the history worth reviewing. Both the resolver and
    ACTIVE_POLICY_HOLDS already filter on isactive, so clearing it is enough to
    resume reconciliation on the next resolve.

    Returns the number of holds released, so a caller can tell "released" from
    "there was nothing to release".
    """
    try:
        from plugins.computers.models import PolicyActionHold
        from shopdb.api import db
    except Exception:
        return 0

    from datetime import datetime
    from sqlalchemy import or_

    # Indefinite holds must be releasable too. Without the null branch the one
    # kind of hold that never clears on its own would also be the one kind the
    # release button could not clear.
    rows = (PolicyActionHold.query
            .filter(PolicyActionHold.assetid == assetid,
                    PolicyActionHold.action == action,
                    PolicyActionHold.isactive.is_(True),
                    or_(PolicyActionHold.expiresat.is_(None),
                        PolicyActionHold.expiresat > datetime.utcnow()))
            .all())
    for row in rows:
        row.isactive = False
    if rows:
        db.session.commit()
    return len(rows)


def ACTIVE_LOCK_FOR(assetid):
    """The endpoint lock standing on this asset, or None.

    Unlike the hold helpers above there is no defensive cross-plugin import:
    endpointlocks belongs to THIS plugin, so if the import could fail the
    enqueue route calling it would not be running either.

    No expiry filter, deliberately. A lock has no TTL - it is cleared by an
    admin or it stands. Anything that looks like an expiry check here would be
    a bug.
    """
    from .models import EndpointLock

    return (EndpointLock.query
            .filter(EndpointLock.assetid == assetid,
                    EndpointLock.releasedat.is_(None))
            .order_by(EndpointLock.lockedat.desc())
            .first())


def PLACE_ENDPOINT_LOCK(assetid, reason, userid=None, username=None):
    """Quarantine one endpoint against non-admin commands.

    Idempotent on the active lock: if one already stands this returns it
    unchanged rather than stacking a second row. Two active locks on one
    endpoint is not a state this table should represent, and the caller who
    "locked" a locked machine got the outcome they asked for either way.

    The invariant is enforced here rather than by a unique constraint because
    released rows accumulate per asset as history, so uniqueness on assetid
    would reject the second lock in an endpoint's life.
    """
    from datetime import datetime

    from shopdb.api import db

    from .models import EndpointLock

    existing = ACTIVE_LOCK_FOR(assetid)
    if existing is not None:
        return existing

    lock = EndpointLock(
        assetid=assetid,
        reason=reason,
        lockedbyuserid=userid,
        lockedbyusername=username,
        lockedat=datetime.utcnow(),
    )
    db.session.add(lock)
    db.session.commit()
    return lock


def RELEASE_ENDPOINT_LOCK(assetid, userid=None, username=None):
    """Lift the quarantine, recording who lifted it.

    Stamps releasedat rather than deleting the row: the lock IS the record that
    this machine was quarantined and why, and that record is most wanted after
    the lock is gone and no longer able to explain itself. Every read filters on
    releasedat, so stamping it is enough to restore normal command flow.

    Returns the released lock, or None when there was nothing to release, so a
    caller can tell those two cases apart.
    """
    from datetime import datetime

    from shopdb.api import db

    lock = ACTIVE_LOCK_FOR(assetid)
    if lock is None:
        return None

    lock.releasedat = datetime.utcnow()
    lock.releasedbyuserid = userid
    lock.releasedbyusername = username
    db.session.commit()
    return lock