admin / Synapse-Cortex

"""Guardrail engine: pure functions, fully unit-testable with no real
infrastructure. Parses a Playbook's visual graph into an ordered command
list, and enforces target-OS and forbidden-command policy before any
command is allowed to reach the execution engine (app/ai/executor.py).

Guardrails here are defense in depth: the tenant's own forbidden-command
list is checked *in addition to* a hardcoded baseline that no tenant
config can override, and the graph itself is re-validated server-side on
every execution attempt (never trusted from a prior save).
"""
import re
from dataclasses import dataclass, field
from typing import Optional


class GuardrailError(ValueError):
    """Raised when a playbook's graph is structurally invalid (can't be
    turned into an execution plan at all) - distinct from a *policy*
    rejection (wrong OS, forbidden command), which is reported as a
    BlockReason instead of raised, since a structurally invalid graph
    should never have been saved and a policy rejection is an expected,
    everyday outcome."""


@dataclass
class PlanStep:
    node_id: str
    label: str
    command: str


# The baseline is split into two tiers. Both are checked in addition to a
# tenant's own forbidden_commands list; patterns are deliberately broad
# (word-boundary / whitespace-tolerant) rather than exact-string, since a
# determined bypass attempt would otherwise just add extra spaces or flags.
#
# Tier 1 - CATASTROPHIC: irreversible data or host destruction. These can
# NEVER be run, acknowledged, or overridden by any tenant/playbook config.
# A match is always a hard block that never reaches human review.
BASELINE_CATASTROPHIC_PATTERNS = [
    (re.compile(r"\brm\s+-[a-z]*r[a-z]*f\b", re.IGNORECASE), "rm -rf"),  # rm -rf, rm -fr, rm -Rf, ...
    (re.compile(r"\bmkfs(\.\w+)?\b", re.IGNORECASE), "mkfs"),
    (re.compile(r"\bdd\b.*\bof=/dev/", re.IGNORECASE), "dd of=/dev/*"),
    (re.compile(r":\(\)\s*\{[^}]*\}\s*;\s*:", re.IGNORECASE), "fork bomb"),  # classic fork bomb
    (re.compile(r">\s*/dev/sd[a-z]\b", re.IGNORECASE), "redirect to a raw disk device"),
    (re.compile(r"\bchmod\s+-R\s+0{3,4}\s+/", re.IGNORECASE), "recursive chmod 000 on /"),
]

# Tier 2 - REVIEW-ELIGIBLE: disruptive but legitimate and reversible actions
# (rebooting or powering off a host). Blocked by default, exactly like the
# catastrophic tier - EXCEPT that a global_admin may "acknowledge" a category
# on a specific playbook (Playbook.acknowledged_dangerous_commands), which
# routes a matching run to MANDATORY human review instead of an auto-block.
# An acknowledged run is never auto-approved; a person still makes the final
# call. Keyed by a stable category id stored on the playbook.
REVIEW_ELIGIBLE_BASELINE: dict[str, dict] = {
    "host_power": {
        "label": "host power control (reboot / shutdown / halt / poweroff)",
        "patterns": [
            re.compile(r"\b(shutdown|reboot|halt|poweroff)\b", re.IGNORECASE),
            re.compile(r"\binit\s+0\b", re.IGNORECASE),
        ],
    },
}

# The set of category ids a playbook may legitimately acknowledge (validated
# at the create/import boundary so a typo never silently does nothing).
ACKNOWLEDGEABLE_CATEGORIES = frozenset(REVIEW_ELIGIBLE_BASELINE)


def parse_graph_to_commands(graph_json: dict) -> list[PlanStep]:
    """Topologically walks the graph from its single required Start node,
    following Condition-node branches (`true` handle preferred, else the
    first available edge), and returns the ordered list of Command steps
    that would execute. Raises GuardrailError for any structural problem:
    no/multiple Start nodes, a cycle, or a Command node unreachable from
    Start (dead code that was probably a mistake, never silently ignored).
    """
    nodes = {n["id"]: n for n in (graph_json or {}).get("nodes", [])}
    edges = (graph_json or {}).get("edges", [])

    start_nodes = [n for n in nodes.values() if n.get("type") == "start"]
    if len(start_nodes) == 0:
        raise GuardrailError("Playbook graph has no Start node.")
    if len(start_nodes) > 1:
        raise GuardrailError("Playbook graph has more than one Start node.")

    outgoing: dict[str, list[dict]] = {}
    for edge in edges:
        outgoing.setdefault(edge["source"], []).append(edge)

    def next_edge(node_id: str) -> Optional[dict]:
        candidates = outgoing.get(node_id, [])
        if not candidates:
            return None
        node = nodes.get(node_id)
        if node and node.get("type") == "condition":
            for handle in ("true", None):
                for edge in candidates:
                    if edge.get("sourceHandle") == handle:
                        return edge
            return candidates[0]
        if len(candidates) > 1:
            raise GuardrailError(f"Node {node_id} has more than one outgoing edge but is not a Condition node.")
        return candidates[0]

    # Pass 1: the single deterministic execution path (preferring a
    # Condition's "true" branch) - this is what actually gets to run.
    steps: list[PlanStep] = []
    path: set[str] = set()

    current_id = start_nodes[0]["id"]
    while current_id is not None:
        if current_id in path:
            raise GuardrailError("Playbook graph contains a cycle.")
        path.add(current_id)

        node = nodes.get(current_id)
        if node is None:
            raise GuardrailError(f"Edge references unknown node {current_id}.")
        if node.get("type") == "command":
            data = node.get("data", {})
            steps.append(PlanStep(node_id=current_id, label=data.get("label", ""), command=data.get("command", "")))

        edge = next_edge(current_id)
        current_id = edge["target"] if edge else None

    # Pass 2: structural reachability from Start over *all* edges (both
    # branches of every Condition), independent of which branch pass 1
    # happened to choose - a Command node only sitting on the "false" side
    # of a condition is legitimately part of the graph, not dead code.
    structurally_reachable: set[str] = set()
    frontier = [start_nodes[0]["id"]]
    while frontier:
        node_id = frontier.pop()
        if node_id in structurally_reachable:
            continue
        structurally_reachable.add(node_id)
        for edge in outgoing.get(node_id, []):
            frontier.append(edge["target"])

    unreachable_commands = [
        n["id"] for n in nodes.values() if n.get("type") == "command" and n["id"] not in structurally_reachable
    ]
    if unreachable_commands:
        raise GuardrailError(f"Playbook graph has unreachable Command node(s): {', '.join(unreachable_commands)}.")

    return steps


def check_target_os(allowed_target_os: list[str], asset_os: Optional[str]) -> Optional[str]:
    """Returns a block reason if the check fails, else None. An empty
    allow-list means no restriction. A non-empty allow-list with no linked
    asset (or no OS recorded on it) always blocks, since there is nothing
    to check the restriction against."""
    if not allowed_target_os:
        return None
    if not asset_os:
        return "Playbook restricts target OS but the ticket's linked asset has no recorded OS."
    lowered = asset_os.lower()
    if any(allowed.lower() in lowered for allowed in allowed_target_os):
        return None
    return f"Asset OS {asset_os!r} does not match the playbook's allowed target OS list ({allowed_target_os!r})."


# Ticket-derived placeholder tokens a Command node's command text may
# contain - inserted via the builder's "+ Username" / "+ IP Address" /
# "+ Password" buttons (CommandNode.tsx) so a template like
# "{{username}}@{{ip_address}}" resolves per-ticket at investigate/execute
# time instead of every playbook needing the host hardcoded.
PLACEHOLDER_USERNAME = "{{username}}"
PLACEHOLDER_IP_ADDRESS = "{{ip_address}}"
PLACEHOLDER_PASSWORD = "{{password}}"
ALL_PLACEHOLDERS = (PLACEHOLDER_USERNAME, PLACEHOLDER_IP_ADDRESS, PLACEHOLDER_PASSWORD)

# The free-form command token. A playbook whose command node contains this is
# a "command runner": the actual command is supplied at run time (proposed by
# the AI investigator, editable by the approving human) rather than fixed in
# the graph. This is the one place an unfixed command enters execution, so the
# substituted result is put through the FULL guardrail check below - crucially
# check_forbidden runs on the substituted command, so the non-overridable
# baseline (rm -rf, mkfs, dd of=/dev, shutdown/reboot, fork bombs, ...) blocks
# a dangerous AI/human command exactly as it would a hardcoded one. Creating
# such a playbook is gated to global_admin and always human-in-the-loop
# (app/routers/playbooks.py), and the substituted command is re-checked again
# at approval time with whatever the human edited it to.
PLACEHOLDER_COMMAND = "{{command}}"


def graph_uses_freeform_command(graph_json: dict) -> bool:
    """True if any command node in the graph uses the {{command}} token -
    scans raw nodes (not the parsed plan) so the global_admin gate applies
    even to a structurally-odd graph."""
    for node in (graph_json or {}).get("nodes", []):
        if PLACEHOLDER_COMMAND in ((node.get("data") or {}).get("command") or ""):
            return True
    return False


def resolve_placeholders(command: str, *, username: Optional[str], ip_address: Optional[str], password: Optional[str]) -> str:
    """Substitutes ticket-derived placeholder tokens into a command string.
    Callers control what `password` resolves to - the caller is responsible
    for passing a masked value (e.g. "********") for anything that gets
    displayed or persisted (proposed plan, execution log), and the real
    decrypted secret only in-memory immediately before dispatching to the
    executor, mirroring app/ai/vault.py's own decrypt-only-immediately-
    before-use discipline. A None value leaves that placeholder untouched
    (used when the caller only wants to resolve a subset)."""
    resolved = command
    if username is not None:
        resolved = resolved.replace(PLACEHOLDER_USERNAME, username)
    if ip_address is not None:
        resolved = resolved.replace(PLACEHOLDER_IP_ADDRESS, ip_address)
    if password is not None:
        resolved = resolved.replace(PLACEHOLDER_PASSWORD, password)
    return resolved


def check_ticket_context(commands: list[str], *, has_username: bool, has_ip: bool, has_password: bool) -> Optional[str]:
    """Blocks execution when a command references ticket data (username,
    IP address, or credential secret) that isn't actually available for
    this ticket yet - a playbook using {{ip_address}} against a ticket with
    no linked asset would otherwise send the literal placeholder text as
    the command."""
    missing = []
    if any(PLACEHOLDER_USERNAME in c for c in commands) and not has_username:
        missing.append("a linked credential (username)")
    if any(PLACEHOLDER_IP_ADDRESS in c for c in commands) and not has_ip:
        missing.append("a linked asset with a recorded IP address")
    if any(PLACEHOLDER_PASSWORD in c for c in commands) and not has_password:
        missing.append("a linked credential (password)")
    if missing:
        return "Playbook references ticket data that isn't available yet: " + ", ".join(missing) + "."
    return None


@dataclass
class ForbiddenResult:
    # A hard block (tenant-forbidden, catastrophic baseline, or an
    # UN-acknowledged review-eligible pattern). None means nothing is blocked.
    block_reason: Optional[str] = None
    # Review-eligible categories that were matched AND acknowledged for this
    # playbook - present but allowed, and the reason the run must go to a human.
    review_categories: list[str] = field(default_factory=list)


def check_forbidden(
    commands: list[str],
    tenant_forbidden: list[str],
    acknowledged: Optional[list[str]] = None,
) -> ForbiddenResult:
    """Checks every command against the tenant's own list (case-insensitive
    substring), the catastrophic baseline (always a hard block), and the
    review-eligible baseline.

    A review-eligible match is a hard block UNLESS its category id is in
    `acknowledged` (a global_admin's per-playbook acknowledgement), in which
    case it is allowed through but recorded in `review_categories` so the
    caller forces the run to mandatory human review. Catastrophic patterns
    can never be acknowledged."""
    ack = set(acknowledged or ())
    review_present: list[str] = []
    for command in commands:
        lowered = command.lower()
        for forbidden in tenant_forbidden:
            if forbidden.lower() in lowered:
                return ForbiddenResult(
                    block_reason=f"Command {command!r} matches tenant-forbidden string {forbidden!r}."
                )
        for pattern, label in BASELINE_CATASTROPHIC_PATTERNS:
            if pattern.search(command):
                return ForbiddenResult(
                    block_reason=(
                        f"Command {command!r} matches a baseline-forbidden pattern ({label}) and can never be run "
                        f"- this category cannot be acknowledged or overridden by any configuration."
                    )
                )
        for category, spec in REVIEW_ELIGIBLE_BASELINE.items():
            if any(p.search(command) for p in spec["patterns"]):
                if category in ack:
                    if category not in review_present:
                        review_present.append(category)
                else:
                    return ForbiddenResult(
                        block_reason=(
                            f"Command {command!r} performs {spec['label']}, which is blocked by default. "
                            f"A global admin can acknowledge '{category}' on this playbook to route it to "
                            f"human review instead of blocking it outright."
                        )
                    )
    return ForbiddenResult(review_categories=review_present)


@dataclass
class GuardrailResult:
    ok: bool
    reason: Optional[str] = None
    steps: Optional[list[PlanStep]] = None
    # True when the plan passed all checks but includes an acknowledged
    # review-eligible command (e.g. a reboot): the caller MUST route the run
    # to human approval and must never auto-approve it. review_reason is a
    # human-readable explanation for the approval card / audit trail.
    requires_review: bool = False
    review_categories: list[str] = field(default_factory=list)
    review_reason: Optional[str] = None


def run_guardrails(
    *,
    graph_json: dict,
    allowed_target_os: list[str],
    forbidden_commands: list[str],
    asset_os: Optional[str],
    has_username: bool = False,
    has_ip: bool = False,
    has_password: bool = False,
    freeform_command: Optional[str] = None,
    acknowledged_dangerous_commands: Optional[list[str]] = None,
) -> GuardrailResult:
    """Orchestrates the full guardrail check. Any failure returns
    ok=False with a specific reason and no steps - the caller must never
    invoke the executor in that case.

    For a command-runner playbook (a command node containing {{command}}),
    `freeform_command` is the run-time command (AI-proposed, human-edited).
    It is substituted into the steps BEFORE the checks run, so the
    non-overridable baseline (check_forbidden) and the OS / ticket-context
    checks all apply to the *actual* command that would execute - a
    dangerous free-form command is blocked exactly like a hardcoded one. The
    returned steps carry the substituted command. A freeform playbook with
    no command supplied is blocked."""
    try:
        steps = parse_graph_to_commands(graph_json)
    except GuardrailError as exc:
        return GuardrailResult(ok=False, reason=str(exc))

    uses_freeform = any(PLACEHOLDER_COMMAND in s.command for s in steps)
    if uses_freeform:
        if not freeform_command or not freeform_command.strip():
            return GuardrailResult(
                ok=False,
                reason="This playbook runs a free-form command, but no command was supplied.",
            )
        steps = [
            PlanStep(
                node_id=s.node_id,
                label=s.label,
                command=s.command.replace(PLACEHOLDER_COMMAND, freeform_command),
            )
            for s in steps
        ]

    commands = [s.command for s in steps]

    reason = check_target_os(allowed_target_os, asset_os)
    if reason:
        return GuardrailResult(ok=False, reason=reason)

    forbidden = check_forbidden(commands, forbidden_commands, acknowledged_dangerous_commands)
    if forbidden.block_reason:
        return GuardrailResult(ok=False, reason=forbidden.block_reason)

    reason = check_ticket_context(commands, has_username=has_username, has_ip=has_ip, has_password=has_password)
    if reason:
        return GuardrailResult(ok=False, reason=reason)

    if forbidden.review_categories:
        labels = [REVIEW_ELIGIBLE_BASELINE[c]["label"] for c in forbidden.review_categories]
        review_reason = (
            "This playbook includes a command that requires explicit human review before it can run: "
            + "; ".join(labels)
            + ". It will never run unattended - a person must approve it."
        )
        return GuardrailResult(
            ok=True,
            steps=steps,
            requires_review=True,
            review_categories=forbidden.review_categories,
            review_reason=review_reason,
        )

    return GuardrailResult(ok=True, steps=steps)