minerops/backend/app/agent_packages/windows.py

"""Build short-lived Windows Agent enrollment packages in memory."""

from __future__ import annotations

import io
import json
import os
import zipfile
from pathlib import Path
from urllib.parse import urlsplit


WINDOWS_PACKAGE_ROOT = "MinerOpsAgent-Windows"

WINDOWS_PACKAGE_FILES = (
    "Enroll-MinerOpsAgent.ps1",
    "Upgrade-MinerOpsAgentCredential.ps1",
    "install.ps1",
    "uninstall.ps1",
    "MinerOpsAgentService.ps1",
    "MinerOpsAgentTray.ps1",
    "MinerOpsAgentGUI.ps1",
    "MinerOpsApi3EnrichmentPoller.ps1",
    "MinerOpsPoolPoller.ps1",
    "MinerOpsPoolPollerTask.ps1",
    "EnrichOneMiner.ps1",
    "MinerOpsControl.ps1",
    "VERSION",
    "README.md",
)

DEFAULT_WINDOWS_AGENT_SOURCE = Path(
    os.environ.get(
        "MINEROPS_WINDOWS_AGENT_SOURCE",
        "/app/agents/windows",
    )
)


class AgentWindowsPackageError(RuntimeError):
    """Base Windows package construction error."""


class AgentWindowsPackageSourceError(AgentWindowsPackageError):
    """Canonical Windows Agent source is unavailable or incomplete."""


def normalize_hub_url(value: str) -> str:
    candidate = value.strip()
    parsed = urlsplit(candidate)

    if parsed.scheme not in {"http", "https"}:
        raise AgentWindowsPackageError(
            "MinerOps Hub URL must use http or https"
        )

    if not parsed.netloc:
        raise AgentWindowsPackageError(
            "MinerOps Hub URL must include a host"
        )

    if parsed.username or parsed.password:
        raise AgentWindowsPackageError(
            "MinerOps Hub URL must not contain credentials"
        )

    if parsed.query or parsed.fragment:
        raise AgentWindowsPackageError(
            "MinerOps Hub URL must not contain query or fragment data"
        )

    if parsed.path not in {"", "/"}:
        raise AgentWindowsPackageError(
            "MinerOps Hub URL must be an origin without a path"
        )

    return f"{parsed.scheme}://{parsed.netloc}"


def build_windows_agent_package(
    *,
    enrollment_token: str,
    hub_url: str,
    workflow: str = "new_agent",
    source_root: Path | None = None,
) -> bytes:
    token = enrollment_token.strip()

    if not 32 <= len(token) <= 256:
        raise AgentWindowsPackageError(
            "Enrollment token has an invalid length"
        )

    normalized_hub_url = normalize_hub_url(hub_url)

    if workflow not in {
        "new_agent",
        "existing_agent_credential_transition",
    }:
        raise AgentWindowsPackageError("Unsupported Windows Agent workflow")

    root = (
        Path(source_root)
        if source_root is not None
        else DEFAULT_WINDOWS_AGENT_SOURCE
    )

    if not root.is_dir():
        raise AgentWindowsPackageSourceError(
            "Canonical Windows Agent source is unavailable"
        )

    missing = [
        name
        for name in WINDOWS_PACKAGE_FILES
        if not (root / name).is_file()
    ]

    if missing:
        raise AgentWindowsPackageSourceError(
            "Canonical Windows Agent source is incomplete: "
            + ", ".join(missing)
        )

    enrollment = {
        "hub_url": normalized_hub_url,
        "token": token,
        "workflow": workflow,
    }

    if workflow == "new_agent":
        command = ".\\Enroll-MinerOpsAgent.ps1"
        description = "enrollment"
    else:
        command = ".\\Upgrade-MinerOpsAgentCredential.ps1"
        description = "existing-Agent credential transition"

    instructions = f"""MinerOps Windows Agent

1. Extract this ZIP on the target Windows machine.
2. Open Windows PowerShell as Administrator.
3. Change to the extracted MinerOpsAgent-Windows folder.
4. Run:

   {command}

This package is for the {description} workflow. The credential is short-lived
and single-use. Do not use the other bootstrap script with this package.
Do not post enrollment.json in chat, tickets, screenshots, or logs.
Delete the ZIP and extracted setup folder after successful installation.
"""

    buffer = io.BytesIO()

    with zipfile.ZipFile(
        buffer,
        mode="w",
        compression=zipfile.ZIP_DEFLATED,
    ) as archive:
        for name in WINDOWS_PACKAGE_FILES:
            archive.write(
                root / name,
                arcname=f"{WINDOWS_PACKAGE_ROOT}/{name}",
            )

        archive.writestr(
            f"{WINDOWS_PACKAGE_ROOT}/enrollment.json",
            json.dumps(
                enrollment,
                indent=2,
                sort_keys=True,
            )
            + "\n",
        )

        archive.writestr(
            f"{WINDOWS_PACKAGE_ROOT}/INSTALL.txt",
            instructions,
        )

    return buffer.getvalue()