PacketStation/server/packetstation/policy/engine.py

"""The policy engine: check one action, or list what a principal can do on a packet.

Order of checks, exactly:
1. removed packet: nothing passes, Admin included.
2. status: the rule's from statuses and the status conditions (RETURN_TO_CLERICAL needs wh_skipped,
   SET_VCLOSED needs routed_to databook). An unknown status fails here.
3. change-order hold, Admin included.
4. system-only rules refuse people; the system principal passes only system-only rules.
5. admin bypass of who and custody, with an IT reason (SET_DATABOOK_AT needs none).
6. filed lock: only SET_DATABOOK_AT and ADMIN_REOPEN, Admin included.
7. who and custody.
8. preconditions in rule order, never bypassed.
"""

from __future__ import annotations

from collections.abc import Callable, Mapping
from dataclasses import dataclass
from types import MappingProxyType
from typing import Any

from ..text import reason_ok
from .actions import EDIT_WHAT, RULES, STATUS_LABELS, STATUSES, Action, Rule
from .custody import allows, holder_name
from .preconditions import INPUT_ONLY, PRECONDITIONS
from .principal import PacketState, Principal

EMPTY: Mapping[str, Any] = MappingProxyType({})

MSG_REMOVED = "This packet was removed."
MSG_UNKNOWN_STATUS = "This packet has an unknown status."
MSG_CO_HOLD = "A change order holds this step."
MSG_SYSTEM_ONLY = "Only the VISUAL feed does this."
MSG_SYSTEM_FORBIDDEN = "The VISUAL feed cannot do this."
MSG_IT_REASON = "Give an IT reason of at least 8 characters."
MSG_FILED = "This packet is filed and locked. IT can reopen it if a correction is needed."
MSG_FORBIDDEN_STEP = "Your access doesn't allow this step."
MSG_NOT_WH_SKIPPED = "This packet came through the warehouse. Send it back to the warehouse instead."
MSG_NOT_DATABOOK = "Only packets routed to the data book are closed in VISUAL here."


@dataclass(frozen=True)
class Decision:
    ok: bool
    code: str = ""  # removed | wrong_status | co_hold | filed | it_reason_required | forbidden | precondition
    #                 | reason_required | system_only
    why: str = ""  # the message for the user
    admin_bypass: bool = False  # an admin passed who and custody only through the bypass (an IT edit)
    signature: str | None = None  # the rule's signature meaning, when one is required
    reasons: tuple[str, ...] = ()  # read mode only: every blocking reason (CO hold and state preconditions)


# Status conditions beyond the from statuses: (predicate, message when it fails).
STATUS_CONDITIONS: dict[Action, tuple[Callable[[PacketState], bool], str]] = {
    Action.RETURN_TO_CLERICAL: (lambda s: s.wh_skipped, MSG_NOT_WH_SKIPPED),
    Action.SET_VCLOSED: (lambda s: s.routed_to == "databook", MSG_NOT_DATABOOK),
}


def _wrong_status(state: PacketState, action: Action, rule: Rule) -> str | None:
    if state.status not in STATUSES:
        return MSG_UNKNOWN_STATUS
    if rule.from_status and state.status not in rule.from_status:
        return f"This step isn't open while the packet is {STATUS_LABELS[state.status]}."
    cond = STATUS_CONDITIONS.get(action)
    if cond is not None and not cond[0](state):
        return cond[1]
    return None


def _forbidden_why(action: Action, rule: Rule, state: PacketState, extra: Mapping[str, Any]) -> str:
    what = EDIT_WHAT.get(action)
    if rule.transition or what is None:
        return MSG_FORBIDDEN_STEP
    holder = extra.get("holder_name")
    if not isinstance(holder, str) or not holder:
        holder = holder_name(state)
    return f"Your access doesn't allow changing {what} on this packet. It's with {holder} right now."


def _evaluate(
    principal: Principal,
    state: PacketState,
    action: Action,
    extra: Mapping[str, Any],
    it_reason: str | None,
    *,
    read: bool,
    assume_it_reason: bool,
) -> Decision:
    rule = RULES[action]
    # (1) removed
    if state.removed:
        return Decision(False, "removed", MSG_REMOVED)
    # (2) status
    why = _wrong_status(state, action, rule)
    if why is not None:
        return Decision(False, "wrong_status", why)
    # (3) change-order hold (deferred in read mode so the UI can show it as blocked after who passes)
    blockers: list[str] = []
    if rule.co_hold and state.co_hold:
        hold_why = state.co_hold_why or MSG_CO_HOLD
        if not read:
            return Decision(False, "co_hold", hold_why)
        blockers.append(hold_why)
    # (4) system
    if rule.system_only and principal.system is None:
        return Decision(False, "system_only", MSG_SYSTEM_ONLY)
    if principal.system is not None and not rule.system_only:
        return Decision(False, "forbidden", MSG_SYSTEM_FORBIDDEN)
    # (5) admin bypass
    who_ok = allows(rule.who, principal, state, extra)
    bypass = False
    if not who_ok and principal.admin:
        if action is Action.SET_DATABOOK_AT:
            bypass = True
        elif (read and assume_it_reason) or reason_ok(it_reason) is not None:
            bypass = True
        else:
            return Decision(False, "it_reason_required", MSG_IT_REASON)
    # (6) filed lock
    if state.status == "filed" and not rule.filed_ok:
        return Decision(False, "filed", MSG_FILED)
    # (7) who and custody
    if not who_ok and not bypass:
        return Decision(False, "forbidden", _forbidden_why(action, rule, state, extra))
    # (8) preconditions, never bypassed
    for name in rule.preconditions:
        if read and name in INPUT_ONLY:
            continue
        failed = PRECONDITIONS[name](state, extra)
        if failed is None:
            continue
        if read:
            blockers.append(failed)
            continue
        code = "reason_required" if rule.reason_required and name == "reason_given" else "precondition"
        return Decision(False, code, failed, admin_bypass=bypass)
    if blockers:
        code = "co_hold" if rule.co_hold and state.co_hold else "precondition"
        return Decision(False, code, blockers[0], admin_bypass=bypass, reasons=tuple(blockers))
    return Decision(True, admin_bypass=bypass, signature=rule.signature)


def check(
    principal: Principal,
    state: PacketState,
    action: Action,
    *,
    extra: Mapping[str, Any] = EMPTY,
    it_reason: str | None = None,
) -> Decision:
    """Decide one action with the request input in `extra`. Never raises for an unknown status."""
    return _evaluate(principal, state, action, extra, it_reason, read=False, assume_it_reason=False)


def evaluate(principal: Principal, state: PacketState, action: Action, *, assume_it_reason: bool = True) -> Decision:
    """Read-mode decision: input-only preconditions skipped, CO hold and state preconditions collected in reasons.

    For an admin the IT reason is assumed when `assume_it_reason` is true; admin_bypass says the UI will ask.
    """
    return _evaluate(principal, state, action, EMPTY, None, read=True, assume_it_reason=assume_it_reason)


def can(
    principal: Principal, state: PacketState, assume_it_reason: bool = True
) -> tuple[list[Action], dict[Action, list[str]]]:
    """What the principal can do now, and what is blocked with the reasons. System-only actions are never listed.

    `can` holds the actions that pass with an empty extra (input-only preconditions skipped). `blocked` holds the
    actions that pass who and custody but fail a state precondition or the CO hold.
    """
    allowed: list[Action] = []
    blocked: dict[Action, list[str]] = {}
    for action, rule in RULES.items():
        if rule.system_only:
            continue
        d = evaluate(principal, state, action, assume_it_reason=assume_it_reason)
        if d.ok:
            allowed.append(action)
        elif d.reasons:
            blocked[action] = list(d.reasons)
    return allowed, blocked