#!/usr/bin/env python3
"""Step-by-step reproduction of the PaperCut Print Deploy client.

This is a guided, interactive re-implementation of the HTTP behaviour observed
in a captured Print Deploy client session.  Every step is announced, executed
against the live server, and validated.  On the first failure the program stops
immediately; after every successful step it asks the user before continuing.

The protocol flow reproduced (in the observed order):

  1. GET  /deploy/config                        Print Deploy server
  2. GET  /brands/auth-dialog-logo.png          optional branding asset
  3. GET  /i18n/en-US                           UI translations
  4. POST /deploy/oauth/url                     start Microsoft sign-in
  5. MANUAL: browser sign-in via localhost:9176 loopback callback
  6. POST /deploy/oauth/user                    exchange code -> session token
  7. GET  /deploy/config                        re-read config after login
  8. POST /deploy/printers                      list printers for this client
  9. GET  /printer-urls                         legacy Mobility Print probe
 10. POST /deploy/oauth/token                   refresh the OAuth tokens
 11. GET  /printer-urls                         legacy probe (again)
 12. GET  /i18n/en-US                           reload translations
 13. GET  /token?printerName=...                Basic auth -> printer JWT
 14. GET  /printer-url?...                      printer queue URL for CUPS
 15. GET  driver zip                            download (optional sha256 check)
 16. GET  catalog extras                        download (optional sha256 check)
 17. GET  capabilities file                     download (optional sha256 check)
 18. GET  defaults file (when advertised)       download (optional sha256 check)

Like the captured client, the driver hashes advertised by /deploy/printers and
driver catalog.json are NOT enforced by default.  Pass --verify-sha256 to make
a mismatch fail the step.

The CUPS/IPP exchange that follows (Create-Job / Send-Document) is not replayed
by default, because it submits a real print job.  If you captured it, pass
--ipp-dump with --submit-print-job to replay those captured IPP requests.

Steps 2, 3, 7, 9, 10, 11 and 12 only observe optional server data and change no
later outcome; --skip-redundant omits them.  Steps 15-18 (the driver downloads)
are omitted by --skip-downloads.  Skipped steps keep their step number in the
output.

Usage:
    python3 -I printdeploy_client.py
    python3 -I printdeploy_client.py --server https://printdeploy.example:9174 \
        --printer Follow-Me --download-dir ./drivers
    python3 -I printdeploy_client.py --skip-redundant --skip-downloads

TLS certificates are not verified by default (the captured client works
against a self-signed PaperCut certificate); pass --verify to enable checks.
"""

from __future__ import annotations

import argparse
import base64
import getpass
import gzip
import hashlib
import http.client
import io
import json
import os
import platform
import posixpath
import re
import secrets
import socket
import ssl
import sys
import threading
import time
import webbrowser
import zipfile
from dataclasses import dataclass, field
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlencode, urlsplit

CLIENT_VERSION = "1.10.3838"
DEFAULT_SERVER = "https://printdeploy.asg.nl:9174"
DEFAULT_USER_AGENT = f"Print Deploy Client/{CLIENT_VERSION}/Debian"
CALLBACK_PATH = "/oauth2/microsoft/callback"


# ---------------------------------------------------------------------------
# Errors and small helpers
# ---------------------------------------------------------------------------


class StepFailed(RuntimeError):
    """A step did not produce the expected result; the program must stop."""


class StepSkipped(RuntimeError):
    """An optional step has nothing to do; continue with the next step."""


def mask(value, keep: int = 8) -> str:
    if not value:
        return "(empty)"
    if len(value) <= keep + 4:
        return value
    return f"{value[:keep]}...[redacted {len(value)} chars]...{value[-4:]}"


_LONG_RUN = re.compile(r"[A-Za-z0-9_\-\.=+/]{80,}")


def snippet(body: bytes, limit: int = 300) -> str:
    text = body[: limit * 4].decode("utf-8", "replace")
    text = _LONG_RUN.sub("<long-value>", text)
    text = " ".join(text.split())
    return text[:limit] + ("..." if len(text) > limit else "")


def human_time(epoch) -> str:
    try:
        return datetime.fromtimestamp(int(epoch), tz=timezone.utc).isoformat()
    except (TypeError, ValueError, OverflowError, OSError):
        return str(epoch)


def b64url_json(part: str):
    part += "=" * (-len(part) % 4)
    return json.loads(base64.urlsafe_b64decode(part.encode("ascii")))


def jwt_summary(token: str) -> str:
    """Describe a JWT without verifying it (metadata only)."""
    try:
        payload = b64url_json(token.split(".")[1])
    except (IndexError, ValueError, UnicodeDecodeError):
        return "not a decodable JWT"
    bits = []
    if "exp" in payload:
        bits.append(f"exp={human_time(payload['exp'])}")
    for key in ("username", "upn", "preferred_username", "sub"):
        if key in payload:
            bits.append(f"{key}={payload[key]}")
            break
    if "scp" in payload:
        bits.append(f"scp={payload['scp']}")
    return ", ".join(bits) or "JWT without exp/username claims"


def build_machine_info(args) -> dict:
    """The client fingerprints the machine; reproduce the observed shape."""
    username = args.username or getpass.getuser()
    hostname = args.hostname or socket.gethostname()
    os_name = args.os_name or platform.system().lower()
    os_version = args.os_version or _distro_version() or platform.release()
    arch = args.arch or _normalize_arch(platform.machine())
    return {
        "version": CLIENT_VERSION,
        "printingSystem": "CUPS",
        "loggedInUser": {"username": username, "type": "LocalMachine"},
        "machine": {
            "hostname": hostname,
            "os": {
                "name": os_name,
                "osName": _distro_name() or platform.system(),
                "version": os_version,
                "arch": arch,
            },
            "ipAddresses": _local_ip_addresses(),
            "activeDirectoryDomainName": "",
        },
    }


def _distro_field(field_name: str):
    try:
        with open("/etc/os-release", encoding="utf-8") as fh:
            for line in fh:
                key, _, value = line.strip().partition("=")
                if key == field_name:
                    return value.strip().strip('"')
    except OSError:
        pass
    return None


def _distro_name():
    name = _distro_field("NAME")
    return re.sub(r"\s+GNU/Linux$", "", name) if name else None


def _distro_version():
    return _distro_field("VERSION_ID")


def _normalize_arch(machine: str) -> str:
    return {
        "x86_64": "amd64",
        "aarch64": "arm64",
        "arm64": "arm64",
        "i686": "386",
        "i386": "386",
    }.get(machine, machine)


def _local_ip_addresses() -> list:
    addresses = set()
    try:
        for info in socket.getaddrinfo(socket.gethostname(), None):
            ip = info[4][0]
            if ":" not in ip and not ip.startswith("127."):
                addresses.add(ip)
    except OSError:
        pass
    if not addresses:
        # UDP connect only selects a route; it sends no packets.
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        try:
            sock.connect(("192.0.2.1", 9))
            ip = sock.getsockname()[0]
            if ip and not ip.startswith("127."):
                addresses.add(ip)
        except OSError:
            pass
        finally:
            sock.close()
    return sorted(addresses)


# ---------------------------------------------------------------------------
# HTTP
# ---------------------------------------------------------------------------


@dataclass
class Response:
    url: str
    status: int
    reason: str
    headers: dict
    body: bytes

    def text(self) -> str:
        return self.body.decode("utf-8", "replace")

    def json(self):
        return json.loads(self.body)


class HttpClient:
    def __init__(self, timeout: float, verify: bool, cacert, verbose: bool, ua: str):
        if cacert:
            self.context = ssl.create_default_context(cafile=cacert)
        elif verify:
            self.context = ssl.create_default_context()
        else:
            self.context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
            self.context.check_hostname = False
            self.context.verify_mode = ssl.CERT_NONE
        self.timeout = timeout
        self.verbose = verbose
        self.user_agent = ua

    def request(self, method: str, url: str, headers=None, body: bytes | None = None):
        parts = urlsplit(url)
        if parts.scheme not in ("http", "https"):
            raise StepFailed(f"unsupported URL scheme in {url!r}")
        port = parts.port or (443 if parts.scheme == "https" else 80)
        if parts.scheme == "https":
            conn = http.client.HTTPSConnection(
                parts.hostname, port, timeout=self.timeout, context=self.context
            )
        else:
            conn = http.client.HTTPConnection(
                parts.hostname, port, timeout=self.timeout
            )
        path = parts.path or "/"
        if parts.query:
            path += "?" + parts.query

        sent = {"User-Agent": self.user_agent, "Accept-Encoding": "gzip"}
        sent.update(headers or {})
        print(f"    -> {method} {url}")
        if self.verbose:
            for name, value in sent.items():
                shown = mask(value) if name.lower() in (
                    "authorization",
                    "proxy-authorization",
                    "cookie",
                ) else value
                print(f"       {name}: {shown}")
            if body is not None:
                print(f"       body: {len(body)} bytes {snippet(body, 160)}")

        try:
            conn.putrequest(method, path, skip_accept_encoding=True)
            for name, value in sent.items():
                conn.putheader(name, value)
            if body is not None:
                conn.putheader("Content-Length", str(len(body)))
            conn.endheaders(body if body else None)
            raw = conn.getresponse()
            data = raw.read()
            response_headers = {k.lower(): v for k, v in raw.getheaders()}
            status, reason = raw.status, raw.reason
        except (OSError, http.client.HTTPException) as exc:
            conn.close()
            raise StepFailed(f"{method} {url} failed: {exc}") from exc
        finally:
            conn.close()

        if response_headers.get("content-encoding", "").lower() == "gzip" and data:
            try:
                data = gzip.decompress(data)
            except OSError:
                pass
        print(f"    <- HTTP {status} {reason} ({len(data)} bytes)")
        return Response(url, status, reason, response_headers, data)


def expect_json(resp: Response, status: int = 200):
    if resp.status != status:
        raise StepFailed(
            f"expected HTTP {status}, got {resp.status} {resp.reason}: {snippet(resp.body)}"
        )
    try:
        return resp.json()
    except ValueError as exc:
        raise StepFailed(f"response is not JSON: {snippet(resp.body)}") from exc


# ---------------------------------------------------------------------------
# OAuth loopback callback
# ---------------------------------------------------------------------------


class _CallbackHandler(BaseHTTPRequestHandler):
    server_version = "PrintDeployRepro/1.0"

    def log_message(self, *args):  # keep the step output clean
        pass

    def do_GET(self):
        parsed = urlsplit(self.path)
        if parsed.path != CALLBACK_PATH:
            self.send_error(404)
            return
        params = parse_qs(parsed.query)
        result = self.server.result
        for key in ("code", "state", "error", "error_description"):
            if key in params:
                result[key] = params[key][0]
        ok = "code" in result
        message = (
            "Sign-in complete. You can close this tab and return to the terminal."
            if ok
            else "Sign-in failed. Check the terminal for details."
        )
        page = f"<html><body><h2>{message}</h2></body></html>".encode()
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(page)))
        self.end_headers()
        self.wfile.write(page)
        self.server.got_response.set()


class _CallbackServer(ThreadingHTTPServer):
    daemon_threads = True
    allow_reuse_address = True

    def server_bind(self):
        if self.address_family == socket.AF_INET6:
            try:
                self.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)
            except OSError:
                pass
        super().server_bind()


def _make_callback_server(port: int) -> _CallbackServer:
    attempts = []
    if socket.has_ipv6:
        attempts.append((socket.AF_INET6, "::"))
    attempts.append((socket.AF_INET, "127.0.0.1"))
    last_error = None
    for family, host in attempts:
        class Server(_CallbackServer):
            address_family = family

        try:
            return Server((host, port), _CallbackHandler)
        except OSError as exc:
            last_error = exc
    raise StepFailed(
        f"cannot listen on port {port} for the OAuth callback ({last_error}). "
        "Close the real Print Deploy client or pass --callback-port."
    )


def wait_for_oauth_callback(port: int, auth_url: str, expected_state: str,
                            timeout: float, open_browser: bool) -> str:
    httpd = _make_callback_server(port)
    httpd.result = {}
    httpd.got_response = threading.Event()
    httpd.timeout = 1

    print(f"    listening on {CALLBACK_PATH} at localhost:{port}")
    print("\n  Open this URL and sign in with the account that has printer access:\n")
    print(f"    {auth_url}\n")
    if open_browser:
        try:
            if webbrowser.open(auth_url):
                print("    (a browser window was opened)")
            else:
                print("    (could not open a browser automatically)")
        except Exception:
            print("    (could not open a browser automatically)")

    deadline = time.monotonic() + timeout
    print(f"    waiting up to {int(timeout)}s for the redirect ...")
    try:
        while not httpd.got_response.is_set() and time.monotonic() < deadline:
            httpd.handle_request()
    finally:
        httpd.server_close()

    if not httpd.got_response.is_set():
        raise StepFailed(
            f"no OAuth callback arrived within {int(timeout)}s; "
            "the sign-in page may have been closed or the account rejected consent"
        )

    result = httpd.result
    if result.get("error"):
        detail = result.get("error_description") or result.get("error")
        raise StepFailed(f"the identity provider returned an error: {detail}")
    code = result.get("code")
    if not code:
        raise StepFailed("the callback did not contain an authorization code")
    if result.get("state") != expected_state:
        raise StepFailed(
            "OAuth state mismatch: the callback state does not match the value "
            "sent to /deploy/oauth/url; aborting to avoid a forged redirect"
        )
    print(f"    <- authorization code received ({len(code)} chars), state matched")
    return code


# ---------------------------------------------------------------------------
# Session state and the step runner
# ---------------------------------------------------------------------------


@dataclass
class Session:
    args: argparse.Namespace
    http: HttpClient
    config: dict = field(default_factory=dict)
    username: str = ""
    session_token: str = ""
    access_token: str = ""
    refresh_token: str = ""
    token_expiry: str = ""
    state: str = ""
    auth_url: str = ""
    auth_code: str = ""
    printers: list = field(default_factory=list)
    printer: dict = field(default_factory=dict)
    mobility: str = ""
    job_token: str = ""
    printer_url: str = ""
    driver_zip_path: str = ""
    catalog_extras: list = field(default_factory=list)

    @property
    def server(self) -> str:
        return self.args.server.rstrip("/")

    @property
    def provider(self) -> str:
        return self.args.provider

    def post_json(self, url: str, payload, headers=None):
        body = json.dumps(payload, separators=(",", ":")).encode()
        return self.http.request("POST", url, headers=headers, body=body)


class Runner:
    def __init__(self):
        self.number = 0

    def step(self, title: str, description: str, action, last: bool = False,
             skip: str = ""):
        self.number += 1
        line = "=" * 72
        print(f"\n{line}\nSTEP {self.number}: {title}\n{line}")
        if skip:
            print(f"[skip] Step {self.number} skipped ({skip}).")
            return None
        print(description.strip())
        print()
        try:
            summary = action()
        except StepSkipped as exc:
            print(f"\n[skip] Step {self.number} skipped: {exc}")
            summary = None
        except StepFailed as exc:
            print(f"\n[FAIL] Step {self.number} failed: {exc}")
            print("Stopping immediately; no further steps were executed.")
            raise SystemExit(1)
        else:
            print(f"\n[ok] Step {self.number} succeeded.")
            if summary:
                for line in str(summary).rstrip().splitlines():
                    print(f"    {line}")
        if not last:
            self.ask_next()
        return summary

    @staticmethod
    def ask_next():
        try:
            answer = input("\nPress Enter to take the next step (q to quit): ")
        except EOFError:
            answer = "q"
        if answer.strip().lower() in ("q", "quit", "exit"):
            print("Stopped at user request. No further steps were executed.")
            raise SystemExit(0)


# ---------------------------------------------------------------------------
# Individual steps
# ---------------------------------------------------------------------------


def step_config(s: Session):
    resp = s.http.request("GET", f"{s.server}/deploy/config")
    cfg = expect_json(resp, 200)
    for key in ("accessibleHost", "AccessibleMobilityPrintTLSPort"):
        if key not in cfg:
            raise StepFailed(f"config response is missing {key!r}")
    s.config = cfg
    return (
        f"accessibleHost = {cfg['accessibleHost']}\n"
        f"Print Deploy TLS port = {cfg.get('accessiblePrintDeployTLSPort')}\n"
        f"Mobility Print TLS port = {cfg['AccessibleMobilityPrintTLSPort']}\n"
        f"auth methods = {cfg.get('AuthMethods')}"
    )


def step_logo(s: Session):
    resp = s.http.request("GET", f"{s.server}/brands/auth-dialog-logo.png")
    if resp.status not in (200, 404):
        raise StepFailed(
            f"branding asset request failed with HTTP {resp.status}: {snippet(resp.body)}"
        )
    if resp.status == 404:
        return "HTTP 404 - this instance has no custom logo (the client tolerates it)"
    return f"HTTP 200 - custom logo fetched ({len(resp.body)} bytes)"


def step_i18n(s: Session):
    resp = s.http.request("GET", f"{s.server}/i18n/en-US")
    strings = expect_json(resp, 200)
    if not isinstance(strings, dict):
        raise StepFailed("translation response is not a JSON object")
    return f"{len(strings)} translation keys loaded"


def step_oauth_url(s: Session):
    s.state = secrets.token_hex(16)
    resp = s.post_json(
        f"{s.server}/deploy/oauth/url",
        {"providerId": s.provider, "state": s.state},
    )
    data = expect_json(resp, 200)
    auth_url = data.get("authURL")
    if not auth_url:
        raise StepFailed("response did not contain authURL")
    query = parse_qs(urlsplit(auth_url).query)
    returned_state = (query.get("state") or [""])[0]
    if returned_state != s.state:
        raise StepFailed(
            "the authURL state does not match the state sent to /deploy/oauth/url"
        )
    redirect = (query.get("redirect_uri") or [""])[0]
    expected_redirect = f"http://localhost:{s.args.callback_port}{CALLBACK_PATH}"
    if redirect != expected_redirect:
        raise StepFailed(
            f"authURL redirects to {redirect!r}, but this program listens on "
            f"{expected_redirect!r}; the instance's OAuth app is configured for a "
            "different callback port, so the code cannot be captured"
        )
    s.auth_url = auth_url
    return f"state = {s.state}\nauthURL = {auth_url}"


def step_login(s: Session):
    s.auth_code = wait_for_oauth_callback(
        s.args.callback_port,
        s.auth_url,
        s.state,
        s.args.login_timeout,
        not s.args.no_browser,
    )
    return f"authorization code captured ({len(s.auth_code)} chars)"


def apply_token(s: Session, token: dict):
    if not isinstance(token, dict):
        raise StepFailed("token object missing from the response")
    access = token.get("access_token")
    if not access:
        raise StepFailed("response did not contain token.access_token")
    s.access_token = access
    s.refresh_token = token.get("refresh_token", s.refresh_token)
    s.token_expiry = token.get("expiry") or human_time(
        time.time() + int(token.get("expires_in") or 0)
    )


def step_oauth_user(s: Session):
    if not s.auth_code:
        raise StepFailed("no authorization code is available")
    resp = s.post_json(
        f"{s.server}/deploy/oauth/user",
        {"providerId": s.provider, "code": s.auth_code},
    )
    data = expect_json(resp, 200)
    for key in ("username", "sessionToken", "token"):
        if key not in data:
            raise StepFailed(f"response is missing {key!r}")
    s.username = data["username"]
    s.session_token = data["sessionToken"]
    apply_token(s, data["token"])
    s.auth_code = ""  # single use: do not keep it around
    return (
        f"username = {s.username}\n"
        f"session token = {mask(s.session_token)}\n"
        f"access token = {mask(s.access_token)}\n"
        f"access token expiry = {s.token_expiry}"
    )


def step_printers(s: Session):
    info = build_machine_info(s.args)
    body = json.dumps(info, separators=(",", ":"))
    resp = s.http.request(
        "POST",
        f"{s.server}/deploy/printers",
        headers={
            "Authorization": f"Bearer {s.session_token}",
            "printdeployauthenticationtype": s.provider,
        },
        body=body.encode(),
    )
    data = expect_json(resp, 200)
    printers = data.get("printers")
    if not isinstance(printers, list) or not printers:
        raise StepFailed("the server returned no printers for this client")
    names = [p.get("name") for p in printers]
    target = next((p for p in printers if p.get("name") == s.args.printer), None)
    if target is None:
        raise StepFailed(
            f"printer {s.args.printer!r} is not offered; available printers: {names}"
        )
    s.printers = printers
    s.printer = target

    connection = target.get("connection") or {}
    base = connection.get("name") or ""
    if not base:
        base = (
            f"https://{s.config['accessibleHost']}:"
            f"{s.config['AccessibleMobilityPrintTLSPort']}"
        )
    if not base.startswith(("http://", "https://")):
        base = "https://" + base
    s.mobility = base.rstrip("/")
    return (
        f"machine fingerprint sent: {body}\n"
        f"printers offered: {names}\n"
        f"selected printer: {target.get('name')!r}\n"
        f"Mobility Print base URL: {s.mobility}"
    )


def step_printer_urls(s: Session):
    resp = s.http.request("GET", f"{s.mobility}/printer-urls")
    if resp.status != 404:
        raise StepFailed(
            f"expected the legacy endpoint to answer 404, got {resp.status}: "
            f"{snippet(resp.body)}"
        )
    return "HTTP 404 - legacy endpoint absent, expected; using per-printer URLs"


def step_refresh_token(s: Session):
    resp = s.post_json(
        f"{s.server}/deploy/oauth/token",
        {},
        headers={"Authorization": f"Bearer {s.session_token}"},
    )
    data = expect_json(resp, 200)
    apply_token(s, data.get("token") or {})
    if data.get("sessionToken"):
        s.session_token = data["sessionToken"]
    return (
        f"refreshed access token = {mask(s.access_token)}\n"
        f"expiry = {s.token_expiry}"
    )


def step_printer_token(s: Session):
    credentials = f"{s.username}:{s.access_token}".encode()
    basic = base64.b64encode(credentials).decode()
    query = urlencode({"printerName": s.args.printer})
    resp = s.http.request(
        "GET",
        f"{s.mobility}/token?{query}",
        headers={
            "Authentication-Type": "openid/microsoft",
            "Authorization": "Basic " + basic,
        },
    )
    data = expect_json(resp, 200)
    token = data.get("token")
    if not token:
        raise StepFailed("response did not contain a printer token")
    s.job_token = token
    return f"printer JWT = {mask(token)}\n{jwt_summary(token)}"


def step_printer_url(s: Session):
    host = urlsplit(s.mobility).hostname
    query = urlencode(
        {"printerName": s.args.printer, "scheme": "http", "serverName": host}
    )
    resp = s.http.request(
        "GET",
        f"{s.mobility}/printer-url?{query}",
        headers={"Authorization": f"Bearer {s.job_token}"},
    )
    if resp.status != 200:
        raise StepFailed(
            f"expected HTTP 200, got {resp.status}: {snippet(resp.body)}"
        )
    url = resp.text().strip()
    if not url.startswith(("http://", "https://")):
        raise StepFailed(f"response is not a printer URL: {url[:200]!r}")
    s.printer_url = url
    return f"printer queue URL = {url}"


def _verify_hashes(s: Session) -> bool:
    return bool(getattr(s.args, "verify_sha256", False))


def _hash_note(expected: str, digest: str, verify: bool) -> str:
    if not expected:
        return f"sha256 {digest} (server advertised no hash)"
    if digest == expected:
        if verify:
            return f"sha256 verified: {digest}"
        return f"sha256 matches server: {digest} (verification disabled)"
    return (
        f"sha256 MISMATCH: got {digest}, server advertised {expected} "
        "(continuing because --verify-sha256 is not set)"
    )


def _download(s: Session, http_path: str, expected_sha: str, dest_name: str):
    if not http_path.startswith("/"):
        http_path = "/" + http_path
    resp = s.http.request("GET", s.server + http_path)
    if resp.status != 200:
        raise StepFailed(
            f"download of {http_path} failed with HTTP {resp.status}: "
            f"{snippet(resp.body)}"
        )
    digest = hashlib.sha256(resp.body).hexdigest()
    verify = _verify_hashes(s)
    if expected_sha and digest != expected_sha:
        if verify:
            raise StepFailed(
                f"sha256 mismatch for {http_path}: expected {expected_sha}, got {digest}"
            )
        print(
            f"    [warn] sha256 mismatch for {http_path}: server advertised "
            f"{expected_sha}, got {digest} (verification disabled, continuing)"
        )
    os.makedirs(s.args.download_dir, exist_ok=True)
    local = os.path.join(s.args.download_dir, dest_name)
    with open(local, "wb") as fh:
        fh.write(resp.body)
    return resp, digest, local


def step_driver_zip(s: Session):
    driver = s.printer.get("driver") or {}
    info = driver.get("zip") or {}
    name = info.get("name")
    if not name:
        raise StepFailed("the selected printer does not advertise a driver zip")
    resp, digest, local = _download(
        s, name, info.get("sha256", ""), f"{s.args.printer}.zip"
    )
    s.driver_zip_path = name
    s.catalog_extras = []
    try:
        with zipfile.ZipFile(io.BytesIO(resp.body)) as archive:
            if "catalog.json" in archive.namelist():
                catalog = json.loads(archive.read("catalog.json").decode("utf-8"))
                s.catalog_extras = catalog.get("files") or []
    except (zipfile.BadZipFile, ValueError, UnicodeDecodeError) as exc:
        raise StepFailed(f"driver zip is not usable: {exc}") from exc
    extra_names = [item.get("name") for item in s.catalog_extras]
    return (
        f"saved {local} ({len(resp.body)} bytes)\n"
        f"{_hash_note(info.get('sha256', ''), digest, _verify_hashes(s))}\n"
        f"catalog extras: {extra_names or 'none'}"
    )


def step_catalog_extras(s: Session):
    if not s.catalog_extras:
        raise StepSkipped("the driver zip declares no extra files")
    base_dir = posixpath.dirname(s.driver_zip_path)
    saved = []
    for item in s.catalog_extras:
        rel = item.get("name")
        if not rel:
            continue
        remote = posixpath.join(base_dir, rel)
        if not remote.startswith("/"):
            remote = "/" + remote
        resp, digest, local = _download(
            s, remote, item.get("sha256", ""), os.path.basename(rel)
        )
        saved.append(
            f"{local} ({len(resp.body)} bytes)\n  "
            f"{_hash_note(item.get('sha256', ''), digest, _verify_hashes(s))}"
        )
    if not saved:
        raise StepSkipped("catalog entries had no usable names")
    return "downloaded catalog extras:\n" + "\n".join(saved)


def step_capabilities(s: Session):
    driver = s.printer.get("driver") or {}
    info = driver.get("capabilities") or {}
    name = info.get("name")
    if not name:
        raise StepFailed("the selected printer does not advertise a capabilities file")
    resp, digest, local = _download(
        s, name, info.get("sha256", ""), f"{s.args.printer}.cap"
    )
    return (
        f"saved {local} ({len(resp.body)} bytes)\n"
        f"{_hash_note(info.get('sha256', ''), digest, _verify_hashes(s))}"
    )


def step_defaults(s: Session):
    driver = s.printer.get("driver") or {}
    info = driver.get("defaults") or {}
    name = info.get("name")
    if not name:
        raise StepSkipped("the printer advertises no defaults file")
    resp, digest, local = _download(
        s, name, info.get("sha256", ""), f"{s.args.printer}.defaults"
    )
    return (
        f"saved {local} ({len(resp.body)} bytes)\n"
        f"{_hash_note(info.get('sha256', ''), digest, _verify_hashes(s))}"
    )


# ---------------------------------------------------------------------------
# Optional: replay captured IPP requests (CUPS print path)
# ---------------------------------------------------------------------------


class _TNError(ValueError):
    pass


def _tn_decode(data: bytes, pos: int):
    colon = data.find(b":", pos)
    if colon < 0:
        raise _TNError(f"missing length at {pos}")
    try:
        length = int(data[pos:colon])
    except ValueError:
        raise _TNError(f"bad length at {pos}") from None
    start = colon + 1
    end = start + length
    if end >= len(data):
        raise _TNError(f"truncated tnetstring at {pos}")
    payload, tag = data[start:end], data[end : end + 1]
    if tag == b",":
        return bytes(payload), end + 1
    if tag == b";":
        return payload.decode("utf-8", "surrogateescape"), end + 1
    if tag == b"#":
        return int(payload), end + 1
    if tag == b"^":
        return float(payload), end + 1
    if tag == b"!":
        return payload == b"true", end + 1
    if tag == b"~":
        return None, end + 1
    if tag == b"}":
        return _tn_dict(payload), end + 1
    if tag == b"]":
        return _tn_list(payload), end + 1
    raise _TNError(f"unknown type {tag!r} at {end}")


def _tn_dict(payload: bytes) -> dict:
    out, pos = {}, 0
    while pos < len(payload):
        key, pos = _tn_decode(payload, pos)
        value, pos = _tn_decode(payload, pos)
        if isinstance(key, bytes):
            key = key.decode("utf-8", "replace")
        out[key] = value
    return out


def _tn_list(payload: bytes) -> list:
    out, pos = [], 0
    while pos < len(payload):
        value, pos = _tn_decode(payload, pos)
        out.append(value)
    return out


def _header_map(msg: dict) -> dict:
    headers = {}
    for item in msg.get("headers") or []:
        if isinstance(item, (list, tuple)) and len(item) == 2:
            key, value = item
            if isinstance(key, bytes):
                key = key.decode("utf-8", "replace")
            if isinstance(value, bytes):
                value = value.decode("utf-8", "replace")
            headers[key.lower()] = value
    return headers


def load_ipp_requests(path: str) -> list:
    with open(path, "rb") as fh:
        data = fh.read()
    out, pos = [], 0
    while pos < len(data):
        if data[pos : pos + 1].isspace():
            pos += 1
            continue
        flow, pos = _tn_decode(data, pos)
        if not isinstance(flow, dict):
            continue
        request = flow.get("request")
        if not isinstance(request, dict):
            continue
        raw_path = request.get("path")
        path_text = raw_path.decode("utf-8", "replace") if isinstance(raw_path, bytes) else str(raw_path or "")
        if not path_text.startswith("/ipp/"):
            continue
        content = request.get("content")
        if isinstance(content, str):
            body = content.encode("utf-8", "surrogateescape")
        elif isinstance(content, (bytes, bytearray)):
            body = bytes(content)
        else:
            continue
        out.append(
            {
                "path": path_text,
                "body": body,
                "headers": _header_map(request),
                "op": int.from_bytes(body[2:4], "big") if len(body) >= 4 else -1,
            }
        )
    return out


IPP_OPS = {
    0x0002: "Print-Job",
    0x0003: "Print-URI",
    0x0004: "Validate-Job",
    0x0005: "Create-Job",
    0x0006: "Send-Document",
    0x0007: "Send-URI",
    0x0008: "Cancel-Job",
    0x0009: "Get-Job-Attributes",
    0x000A: "Get-Jobs",
    0x000B: "Get-Printer-Attributes",
}

IPP_STATUS = {
    0x0000: "successful-ok",
    0x0001: "successful-ok-ignored-or-substituted-attributes",
    0x0400: "client-error-bad-request",
    0x0401: "client-error-forbidden",
    0x0402: "client-error-not-authenticated",
    0x0403: "client-error-not-authorized",
    0x0404: "client-error-not-possible",
    0x0405: "client-error-timeout",
    0x0406: "client-error-not-found",
    0x0407: "client-error-gone",
    0x0500: "server-error-internal-error",
    0x0501: "server-error-operation-not-supported",
}


def _ipp_attributes(data: bytes) -> dict:
    """Parse the top-level IPP attributes of a response (enough for job-id)."""
    out: dict[str, list] = {}
    pos = 8
    last_name = None
    while pos < len(data):
        tag = data[pos]
        if tag == 0x03:  # end-of-attributes
            break
        if tag < 0x10:  # group delimiter
            pos += 1
            continue
        if pos + 3 > len(data):
            break
        name_length = int.from_bytes(data[pos + 1 : pos + 3], "big")
        name_bytes = data[pos + 3 : pos + 3 + name_length]
        pos += 3 + name_length
        name = name_bytes.decode("latin-1") if name_bytes else last_name
        if name_bytes:
            last_name = name
        if pos + 2 > len(data):
            break
        value_length = int.from_bytes(data[pos : pos + 2], "big")
        value = data[pos + 2 : pos + 2 + value_length]
        pos += 2 + value_length
        if name:
            out.setdefault(name, []).append((tag, value))
    return out


_JOB_ID_PREFIX = b"\x21\x00\x06job-id\x00\x04"


def _rewrite_job_id(body: bytes, job_id: int) -> tuple[bytes, bool]:
    index = body.find(_JOB_ID_PREFIX)
    if index < 0:
        return body, False
    value_at = index + len(_JOB_ID_PREFIX)
    return body[:value_at] + job_id.to_bytes(4, "big") + body[value_at + 4 :], True


JOB_DEPENDENT_OPS = {0x0002, 0x0003, 0x0005, 0x0006, 0x0007, 0x0008, 0x0009}


def prepare_ipp_steps(s: Session, dump_path: str):
    requests = load_ipp_requests(dump_path)
    if not requests:
        raise StepFailed(f"no IPP requests found in {dump_path}")

    captured_printer = requests[0]["path"].rstrip("/").split("/")[-1]
    if captured_printer != s.args.printer:
        raise StepFailed(
            f"{dump_path} targets printer {captured_printer!r}, but this run "
            f"selected {s.args.printer!r}; refusing to replay print traffic for "
            "a different printer"
        )

    has_job = any(req["op"] in JOB_DEPENDENT_OPS for req in requests)
    include_jobs = False
    if has_job:
        print(
            "\nThe capture contains print-job requests (Create-Job / Send-Document).\n"
            "Replaying them submits a real print job (the captured document) to "
            "the printer.\n"
        )
        if not s.args.submit_print_job:
            print(
                "Print-job requests will be skipped. Re-run with "
                "--submit-print-job to include them."
            )
        else:
            answer = input(
                "Type PRINT to include the print-job requests, anything else to skip: "
            )
            include_jobs = answer.strip() == "PRINT"
            if not include_jobs:
                print("Print-job requests skipped.")

    return [
        req
        for req in requests
        if include_jobs or req["op"] not in JOB_DEPENDENT_OPS
    ]


def replay_ipp(s: Session, runner: Runner, requests: list):
    job_id = None
    for index, request in enumerate(requests):
        op = request["op"]
        op_name = IPP_OPS.get(op, f"operation-0x{op:04x}")
        description = (
            "Replaying a captured IPP request from the client to the printer.\n"
            f"operation: {op_name}\n"
            f"path: {request['path']}\n"
            f"request body: {len(request['body'])} bytes"
        )
        if op == 0x0006:
            description += (
                "\nThis request carries the document data; executing it submits "
                "the captured print job."
            )

        def action(request=request, op=op, op_name=op_name):
            nonlocal job_id
            body = request["body"]
            if op in (0x0006, 0x0009) and job_id is not None:
                body, patched = _rewrite_job_id(body, job_id)
                if not patched:
                    raise StepFailed(
                        f"captured {op_name} request has no job-id attribute to "
                        "rewrite for the newly created job"
                    )
            headers = {
                "Content-Type": "application/ipp",
                "Authorization": f"Bearer {s.job_token}",
            }
            forwarded = request["headers"].get("x-forwarded-for")
            if forwarded:
                headers["X-Forwarded-For"] = forwarded
            resp = s.http.request(
                "POST",
                s.mobility + request["path"],
                headers=headers,
                body=body,
            )
            if resp.status != 200:
                raise StepFailed(
                    f"IPP {op_name} returned HTTP {resp.status}: {snippet(resp.body)}"
                )
            status = (
                int.from_bytes(resp.body[2:4], "big") if len(resp.body) >= 4 else -1
            )
            if status != 0:
                raise StepFailed(
                    f"IPP {op_name} failed: 0x{status:04x} "
                    f"({IPP_STATUS.get(status, 'unknown')})"
                )
            if op == 0x0005:  # Create-Job: remember the new job id
                values = _ipp_attributes(resp.body).get("job-id")
                if not values:
                    raise StepFailed(
                        "Create-Job succeeded but the response has no job-id; the "
                        "captured job-attribute requests cannot be retargeted"
                    )
                job_id = int.from_bytes(values[0][1], "big")
                return (
                    f"IPP status 0x0000 ({IPP_STATUS[0]}), job-id={job_id}, "
                    f"{len(resp.body)} response bytes"
                )
            return (
                f"IPP status 0x0000 ({IPP_STATUS[0]}), "
                f"{len(resp.body)} response bytes"
            )

        runner.step(
            f"IPP {op_name}", description, action, last=index == len(requests) - 1
        )


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="Interactively reproduce the PaperCut Print Deploy client "
        "protocol, one step at a time.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    p.add_argument("--server", default=DEFAULT_SERVER,
                   help="Print Deploy server base URL")
    p.add_argument("--printer", default="Follow-Me",
                   help="printer name to select from /deploy/printers")
    p.add_argument("--provider", default="microsoft",
                   help="identity provider id for the OAuth flow")
    p.add_argument("--callback-port", type=int, default=9176,
                   help="local port for the OAuth loopback callback")
    p.add_argument("--login-timeout", type=float, default=300,
                   help="seconds to wait for the browser sign-in callback")
    p.add_argument("--no-browser", action="store_true",
                   help="do not open a browser automatically")
    p.add_argument("--username", help="username sent in the machine fingerprint")
    p.add_argument("--hostname", help="hostname sent in the machine fingerprint")
    p.add_argument("--os-name", dest="os_name",
                   help="os.name in the machine fingerprint (e.g. darwin)")
    p.add_argument("--os-version", dest="os_version",
                   help="os.version in the machine fingerprint")
    p.add_argument("--arch", help="os.arch in the machine fingerprint")
    p.add_argument("--download-dir", default="printdeploy_downloads",
                   help="directory for downloaded driver files")
    p.add_argument("--skip-redundant", action="store_true",
                   help="skip steps 2, 3, 7, 9, 10, 11 and 12: branding, the "
                        "translation loads, the repeated config fetch, the "
                        "legacy /printer-urls probes and the token refresh. "
                        "The refresh can renew an expired access token, so "
                        "skip it only while the sign-in token is still valid")
    p.add_argument("--skip-downloads", action="store_true",
                   help="skip steps 15-18: the driver zip, catalog extras, "
                        "capabilities and defaults downloads")
    p.add_argument("--verify-sha256", action="store_true",
                   help="fail a download step when its sha256 does not match the "
                        "hash advertised by the server (default: off, like the "
                        "captured client)")
    p.add_argument("--verify", action="store_true",
                   help="verify TLS certificates (off by default)")
    p.add_argument("--cacert", help="CA bundle to trust (implies verification)")
    p.add_argument("--timeout", type=float, default=30.0,
                   help="per-request network timeout in seconds")
    p.add_argument("--user-agent", default=DEFAULT_USER_AGENT)
    p.add_argument("--verbose", action="store_true",
                   help="show request headers and bodies (secrets masked)")
    p.add_argument("--ipp-dump", metavar="DUMP",
                   help="after the HTTP steps, replay IPP requests captured in "
                        "this mitmproxy flow dump")
    p.add_argument("--submit-print-job", action="store_true",
                   help="with --ipp-dump, also replay Create-Job/Send-Document "
                        "(submits a real print job)")
    return p


def main(argv=None) -> int:
    args = build_parser().parse_args(argv)
    args.server = args.server.rstrip("/")

    # Keep step output visible when stdout is piped or logged to a file.
    try:
        sys.stdout.reconfigure(line_buffering=True)
    except (AttributeError, ValueError):
        pass

    print("PaperCut Print Deploy client - step-by-step reproduction")
    print(f"  server:  {args.server}")
    print(f"  printer: {args.printer}")
    if args.skip_redundant:
        print("  redundant steps:  skipped (--skip-redundant)")
    if args.skip_downloads:
        print("  driver downloads: skipped (--skip-downloads)")
    print(f"  TLS verification: {'on' if (args.verify or args.cacert) else 'OFF'}")
    print(f"  driver sha256:    {'verified (--verify-sha256)' if args.verify_sha256 else 'not enforced'}")
    if not (args.verify or args.cacert):
        print("  (use --verify or --cacert if the server has a trusted certificate)")

    http = HttpClient(
        timeout=args.timeout,
        verify=args.verify,
        cacert=args.cacert,
        verbose=args.verbose,
        ua=args.user_agent,
    )
    session = Session(args=args, http=http)
    runner = Runner()
    redundant = "--skip-redundant" if args.skip_redundant else ""
    downloads = "--skip-downloads" if args.skip_downloads else ""

    runner.step(
        "Fetch the Print Deploy configuration",
        "GET /deploy/config discovers the Mobility Print endpoint, the auth "
        "methods and the polling configuration.",
        lambda: step_config(session),
    )
    runner.step(
        "Fetch the sign-in dialog branding",
        "The client asks for a branded logo shown in the sign-in dialog. "
        "A 404 is normal when the instance has no custom branding.",
        lambda: step_logo(session),
        skip=redundant,
    )
    runner.step(
        "Fetch UI translations",
        "GET /i18n/en-US loads the strings shown by the client.",
        lambda: step_i18n(session),
        skip=redundant,
    )
    runner.step(
        "Start the Microsoft sign-in",
        "POST /deploy/oauth/url asks the server for the Microsoft authorization "
        "URL. A fresh random state is generated for this run.",
        lambda: step_oauth_url(session),
    )
    runner.step(
        "Sign in with Microsoft in a browser",
        "The instance's OAuth app redirects to the loopback URL "
        f"http://localhost:{args.callback_port}{CALLBACK_PATH}.\n"
        "This step starts a temporary listener there, opens the sign-in page, "
        "and captures the authorization code from the redirect.",
        lambda: step_login(session),
    )
    runner.step(
        "Exchange the authorization code for a session",
        "POST /deploy/oauth/user sends the code and receives the PaperCut "
        "session token plus the Microsoft access/refresh tokens.",
        lambda: step_oauth_user(session),
    )
    runner.step(
        "Re-read the configuration after sign-in",
        "The observed client fetches /deploy/config again once the user is "
        "signed in.",
        lambda: step_config(session),
        skip=redundant,
    )
    runner.step(
        "List printers for this machine",
        "POST /deploy/printers authenticates with the session token and sends "
        "a machine fingerprint. The response contains the drivers to install.",
        lambda: step_printers(session),
    )
    runner.step(
        "Probe the legacy /printer-urls endpoint",
        "The client checks the old Mobility Print endpoint first. A 404 is "
        "expected on current servers and the client falls back to per-printer "
        "URLs.",
        lambda: step_printer_urls(session),
        skip=redundant,
    )
    runner.step(
        "Refresh the OAuth tokens",
        "POST /deploy/oauth/token uses the session token to obtain a fresh "
        "Microsoft access token.",
        lambda: step_refresh_token(session),
        skip=redundant,
    )
    runner.step(
        "Probe the legacy /printer-urls endpoint again",
        "The client repeats the legacy probe after refreshing its tokens.",
        lambda: step_printer_urls(session),
        skip=redundant,
    )
    runner.step(
        "Reload UI translations",
        "The client reloads /i18n/en-US before setting up the printer.",
        lambda: step_i18n(session),
        skip=redundant,
    )
    runner.step(
        "Request the printer JWT",
        "GET /token authenticates with HTTP Basic using "
        "username:Microsoft-access-token and the Authentication-Type header. "
        "The response is the JWT used for the printer queue.",
        lambda: step_printer_token(session),
    )
    runner.step(
        "Resolve the printer queue URL",
        "GET /printer-url turns the printer JWT into the queue URL that the "
        "client hands to CUPS.",
        lambda: step_printer_url(session),
    )
    runner.step(
        "Download the printer driver package",
        "The driver zip advertised by /deploy/printers is downloaded. Its "
        "sha256 is only enforced with --verify-sha256. catalog.json lists any "
        "extra files.",
        lambda: step_driver_zip(session),
        skip=downloads,
    )
    runner.step(
        "Download the driver catalog extras",
        "Files listed in the driver zip's catalog.json are fetched relative to "
        "the driver path. Their sha256 is only enforced with --verify-sha256.",
        lambda: step_catalog_extras(session),
        skip=downloads,
    )
    runner.step(
        "Download the printer capabilities",
        "The capabilities file from /deploy/printers is downloaded. Its sha256 "
        "is only enforced with --verify-sha256.",
        lambda: step_capabilities(session),
        skip=downloads,
    )
    runner.step(
        "Download the printer defaults (if advertised)",
        "Some drivers ship a defaults file; it is downloaded when the printer "
        "response advertises one. Its sha256 is only enforced with "
        "--verify-sha256.",
        lambda: step_defaults(session),
        last=not args.ipp_dump,
        skip=downloads,
    )

    if args.ipp_dump:
        try:
            ipp_requests = prepare_ipp_steps(session, args.ipp_dump)
        except (StepFailed, OSError, ValueError) as exc:
            print(f"\n[FAIL] Cannot prepare the IPP replay: {exc}")
            return 1
        replay_ipp(session, runner, ipp_requests)
    else:
        print(
            "\nThe captured client then configures a CUPS queue and speaks IPP "
            "to the printer\n"
            "(Get-Printer-Attributes / Create-Job / Send-Document). That phase "
            "submits a real\n"
            "print job, so it is not repeated by default. To replay it from the "
            "capture:\n"
            "    --ipp-dump convo6_reallyfull.dump [--submit-print-job]\n"
        )

    print("\nAll steps completed.")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except KeyboardInterrupt:
        print("\nInterrupted.")
        raise SystemExit(130)