draupnir 0.1.9

//! **Draupnir's own BMC** — a pure-Rust **Redfish server** that fronts KVM.
//!
//! `src/redfish.rs` is the Redfish **client**: it drives a real iDRAC / iLO /
//! OpenBMC. This is the other end — a Redfish **service** that answers the same
//! endpoints and, instead of a baseboard controller wired to a motherboard, drives
//! a QEMU/KVM guest through draupnir's own [`Boot`] / [`Lifecycle`] seam.
//!
//! # Why it exists
//!
//! Two reasons, and the second is the durable one.
//!
//! 1. A Redfish burn needs a **BMC**: the always-on management controller a real
//!    server has and a developer box does not. Without one the `iso-metal` proof is
//!    a human with a USB stick, which is not a matrix row. This gives the wire half
//!    of that proof a service to run against.
//! 2. A Rust Redfish server fronting KVM turns **any Linux box into a
//!    Redfish-controllable machine**. That is squarely this crate's charter (kvm and
//!    container, low level), and it is useful far outside a test harness: a lab, a
//!    CI fleet, a nested-virt pool can all be addressed with the same out-of-band
//!    vocabulary as the metal they stand in for.
//!
//! # What it does NOT prove — read this before believing a green
//!
//! **An emulator proves the wire, never the hardware.** draupnir's client talking to
//! draupnir's server is a mirror: both ends can be wrong in the same way and every
//! round trip still passes. A loopback green on its own is evidence of
//! self-consistency and of nothing else.
//!
//! Two things are done about that, and neither is optional:
//!
//! * **The shapes are anchored outside this repo.** `tests/redfish_server_conformance.rs`
//!   checks every response against **DMTF's own published JSON Schema, message
//!   registry and mockup** (vendored under `tests/fixtures/dmtf/`, provenance and
//!   hashes in `PROVENANCE.md`). Required properties, `@odata.type` versions,
//!   `@Redfish.AllowableValues` subsets and the `error` payload are all read out of
//!   DMTF's files, not out of ours.
//! * **The result is recorded as `iso-redfish-sim`, never `iso-metal`.** Recording an
//!   emulator's green in the cell a real-hardware green would occupy is exactly the
//!   false green the form split exists to prevent. `iso-metal` stays Absent.
//!
//! # The translation
//!
//! Every Redfish verb lands on vocabulary this crate already had — nothing new was
//! invented for the server side:
//!
//! | Redfish | draupnir |
//! |---|---|
//! | `VirtualMedia.InsertMedia` | [`BootSpec::medium`] |
//! | `Boot.BootSourceOverrideTarget` | [`BootOrder`] (via [`BootOrder::from_redfish_target`]) |
//! | `ComputerSystem.Reset` | [`Boot::boot`] / [`Lifecycle::power_off`] |
//! | `ComputerSystem.PowerState` | [`Lifecycle::status`] |
//!
//! and the URLs, property names and enum tokens come from [`crate::redfish::wire`],
//! the **one** module both ends read. The client composes `host + path()`; the
//! server routes on `path()`. They agree by construction.
//!
//! # Security posture
//!
//! draupnir's client **pins** the BMC certificate, so this server needs a real TLS
//! identity to be pinned. It mints a genuine self-signed X.509 with SANs per run
//! (`rcgen` over `ring` — the same crate, features and call shape as Skidbladnir's
//! `pki::issue_leaf`) and holds the key **in memory only**: nothing is written to
//! disk, so there is no `.key`/`.pem`/`.der` that could be committed. The cert PEM is
//! handed out through [`RedfishKvmServer::cert_pem`] for the client to pin.
//!
//! Everything except the protocol-version probe and the service root requires HTTP
//! Basic auth, which is what DSP0266 permits an unauthenticated client to reach.
//!
//! # Example
//!
//! ```no_run
//! use draupnir::redfish_server::{NodeConfig, RedfishKvmServer};
//! use draupnir::redfish::RedfishBoot;
//! use draupnir::{Boot, BootSpec};
//!
//! // A BMC in front of this box's KVM.
//! let server = RedfishKvmServer::start(
//!     NodeConfig::new("System.Embedded.1")
//!         .credentials("admin", "secret")
//!         .local_disk("/var/lib/node.qcow2"),
//! )?;
//!
//! // ...driven by draupnir's ordinary Redfish CLIENT, cert pinned.
//! let backend = RedfishBoot::new()
//!     .with_password("secret")
//!     .pin_cert_pem(server.cert_pem().as_bytes().to_vec());
//! let spec = BootSpec::iso_boot("burn", "/images/appliance.iso")
//!     .on_metal(server.bmc_endpoint());
//! let machine = backend.boot(&spec)?;
//! # Ok::<(), draupnir::Error>(())
//! ```

use std::collections::BTreeMap;
use std::io::{Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use serde_json::{json, Value};

use crate::kvm::KvmBoot;
use crate::redfish::wire::{self, prop};
use crate::{
    BmcEndpoint, Boot, BootOrder, BootSpec, BootTarget, Error, Lifecycle, Machine, PowerState,
    Result, Seen,
};

// ===========================================================================
// DMTF message registry — the error vocabulary, transcribed and then CHECKED
// ===========================================================================

/// The DMTF **Base** message registry this service reports `MessageId`s from.
///
/// Every `MessageId` it emits is `{BASE_REGISTRY}.{Id}`, and every message body is
/// the registry's own template with `MessageArgs` substituted. The conformance test
/// loads `tests/fixtures/dmtf/registries/Base.1.19.0.json` and asserts that each
/// [`Message`] below matches the registry's `Message`, `MessageSeverity`,
/// `NumberOfArgs` and `Resolution` **exactly** — so a hand-typed error string that
/// merely *sounds* like Redfish cannot survive here.
pub const BASE_REGISTRY: &str = "Base.1.19.0";

/// One entry of the DMTF Base registry, as this service reports it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Message {
    /// The registry entry's `Id` (the part after the registry name in a `MessageId`).
    pub id: &'static str,
    /// The registry's `Message` template, with `%1`..`%N` placeholders.
    pub template: &'static str,
    /// The registry's `MessageSeverity`.
    pub severity: &'static str,
    /// The registry's `Resolution`.
    pub resolution: &'static str,
    /// The registry's `NumberOfArgs`.
    pub nargs: usize,
    /// The HTTP status this service answers with when it reports this message.
    pub status: u16,
}

impl Message {
    /// The full `MessageId`, e.g. `Base.1.19.0.ActionParameterValueNotInList`.
    pub fn message_id(&self) -> String {
        format!("{BASE_REGISTRY}.{}", self.id)
    }

    /// Render the registry template with `args` substituted for `%1`..`%N`.
    ///
    /// A wrong arity is a **programming** error in this crate, not a client error, so
    /// it is caught by a debug assertion and then rendered honestly (the unfilled
    /// placeholder stays visible) rather than papered over.
    pub fn render(&self, args: &[&str]) -> String {
        debug_assert_eq!(
            args.len(),
            self.nargs,
            "{} takes {} args, got {}",
            self.id,
            self.nargs,
            args.len()
        );
        let mut out = self.template.to_string();
        // Descending index, so `%10` (were the registry ever to grow one) is not
        // eaten by `%1`.
        for (i, a) in args.iter().enumerate().rev() {
            out = out.replace(&format!("%{}", i + 1), a);
        }
        out
    }
}

/// `The value '%1' for the parameter %2 in the action %3 is not in the list of
/// acceptable values.` — a bad `ResetType`.
pub const ACTION_PARAMETER_VALUE_NOT_IN_LIST: Message = Message {
    id: "ActionParameterValueNotInList",
    template: "The value '%1' for the parameter %2 in the action %3 is not in the list of acceptable values.",
    severity: "Warning",
    resolution: "Choose a value from the enumeration list that the implementation can support and resubmit the request if the operation failed.",
    nargs: 3,
    status: 400,
};

/// `The action %1 requires the parameter %2 to be present in the request body.`
pub const ACTION_PARAMETER_MISSING: Message = Message {
    id: "ActionParameterMissing",
    template: "The action %1 requires the parameter %2 to be present in the request body.",
    severity: "Critical",
    resolution: "Supply the action with the required parameter in the request body when the request is resubmitted.",
    nargs: 2,
    status: 400,
};

/// `The value '%1' for the property %2 is not in the list of acceptable values.` —
/// a `BootSourceOverrideTarget` this node cannot honour.
pub const PROPERTY_VALUE_NOT_IN_LIST: Message = Message {
    id: "PropertyValueNotInList",
    template: "The value '%1' for the property %2 is not in the list of acceptable values.",
    severity: "Warning",
    resolution: "Choose a value from the enumeration list that the implementation can support and resubmit the request if the operation failed.",
    nargs: 2,
    status: 400,
};

/// `The resource at the URI '%1' was not found.` — an `InsertMedia` pointing at an
/// image that is not there. THE red control: a media URI nobody built must be
/// refused **by name**, not mounted as an empty tray.
pub const RESOURCE_MISSING_AT_URI: Message = Message {
    id: "ResourceMissingAtURI",
    template: "The resource at the URI '%1' was not found.",
    severity: "Critical",
    resolution: "Place a valid resource at the URI or correct the URI and resubmit the request.",
    nargs: 1,
    status: 400,
};

/// `The requested resource of type %1 named '%2' was not found.`
pub const RESOURCE_NOT_FOUND: Message = Message {
    id: "ResourceNotFound",
    template: "The requested resource of type %1 named '%2' was not found.",
    severity: "Critical",
    resolution: "Provide a valid resource identifier and resubmit the request.",
    nargs: 2,
    status: 404,
};

/// `There is no valid session established with the implementation.` — no (or bad)
/// credentials.
pub const NO_VALID_SESSION: Message = Message {
    id: "NoValidSession",
    template: "There is no valid session established with the implementation.",
    severity: "Critical",
    resolution: "Establish a session before attempting any operations.",
    nargs: 0,
    status: 401,
};

/// `The action %1 is not supported by the resource.`
pub const ACTION_NOT_SUPPORTED: Message = Message {
    id: "ActionNotSupported",
    template: "The action %1 is not supported by the resource.",
    severity: "Critical",
    resolution: "Check the Actions property in the resource for the supported actions.",
    nargs: 1,
    status: 400,
};

/// `The request body submitted was malformed JSON and could not be parsed by the
/// receiving service.`
pub const MALFORMED_JSON: Message = Message {
    id: "MalformedJSON",
    template: "The request body submitted was malformed JSON and could not be parsed by the receiving service.",
    severity: "Critical",
    resolution: "Ensure that the request body is valid JSON and resubmit the request.",
    nargs: 0,
    status: 400,
};

/// `A general error has occurred.  See Resolution for information on how to resolve
/// the error, or @Message.ExtendedInfo if Resolution is not provided.` — the
/// catch-all for a backend that refused (QEMU would not start, no medium inserted,
/// no local disk configured). The **detail always names what happened**, in the
/// `Resolution`-shaped second sentence of the extended info.
pub const GENERAL_ERROR: Message = Message {
    id: "GeneralError",
    template: "A general error has occurred.  See Resolution for information on how to resolve the error, or @Message.ExtendedInfo if Resolution is not provided.",
    severity: "Critical",
    resolution: "None.",
    nargs: 0,
    status: 500,
};

/// Every message this service can emit — the list the conformance test walks when it
/// checks each one against DMTF's registry. A new message added below and forgotten
/// here is a message nothing checks, so this array is the gate.
pub const ALL_MESSAGES: &[Message] = &[
    ACTION_PARAMETER_VALUE_NOT_IN_LIST,
    ACTION_PARAMETER_MISSING,
    PROPERTY_VALUE_NOT_IN_LIST,
    RESOURCE_MISSING_AT_URI,
    RESOURCE_NOT_FOUND,
    NO_VALID_SESSION,
    ACTION_NOT_SUPPORTED,
    MALFORMED_JSON,
    GENERAL_ERROR,
];

/// The `@odata.type` values this service reports, one per resource. Pinned as
/// constants because the conformance test resolves each against the matching DMTF
/// schema file and fails if the named version does not define the named type.
mod odata_type {
    /// `ServiceRoot`.
    pub const SERVICE_ROOT: &str = "#ServiceRoot.v1_16_1.ServiceRoot";
    /// `ComputerSystemCollection` (collections are unversioned).
    pub const SYSTEM_COLLECTION: &str = "#ComputerSystemCollection.ComputerSystemCollection";
    /// `ComputerSystem`.
    pub const SYSTEM: &str = "#ComputerSystem.v1_22_0.ComputerSystem";
    /// `VirtualMediaCollection` (unversioned).
    pub const VIRTUAL_MEDIA_COLLECTION: &str = "#VirtualMediaCollection.VirtualMediaCollection";
    /// `VirtualMedia`.
    pub const VIRTUAL_MEDIA: &str = "#VirtualMedia.v1_6_3.VirtualMedia";
    /// `SessionCollection` (unversioned).
    pub const SESSION_COLLECTION: &str = "#SessionCollection.SessionCollection";
}

/// The Redfish protocol version this service reports as `ServiceRoot.RedfishVersion`.
const REDFISH_VERSION: &str = "1.15.0";

// ===========================================================================
// Configuration
// ===========================================================================

/// **The node this BMC fronts** — what the emulated `ComputerSystem` looks like and
/// which local resources it actually drives.
#[derive(Debug, Clone)]
pub struct NodeConfig {
    /// The Redfish `ComputerSystem` id — the `{system_id}` in every URL, and what a
    /// [`BmcEndpoint`] carries. Choose the same spelling a real BMC of the fleet
    /// would use (`System.Embedded.1` for iDRAC, `1` for iLO) so a spec written
    /// against the emulator is byte-identical against the metal.
    pub system_id: String,
    /// The `VirtualMedia` slot id the ISO is inserted into. Defaults to
    /// [`crate::redfish::DEFAULT_MEDIA_ID`] (`"CD"`) — the slot draupnir's own client
    /// targets when nothing else is said. (DMTF's mockup happens to call its slot
    /// `CD1`, and iLO calls it `1`: the id is an instance detail, never spec-fixed.)
    pub media_slot: String,
    /// The Redfish account username.
    pub username: String,
    /// The Redfish account password.
    pub password: String,
    /// The address to bind. `127.0.0.1:0` (the default) takes an ephemeral port;
    /// read the real one back with [`RedfishKvmServer::addr`].
    pub bind: SocketAddr,
    /// **The node's local disk** — what a `BootSourceOverrideTarget: "Hdd"` actually
    /// boots. `None` means this node has no disk, and an `Hdd` override is then
    /// refused by name rather than silently falling through to the medium (which is
    /// precisely how a boot override gets "accepted" without ever being applied).
    pub local_disk: Option<String>,
    /// Guest RAM in MiB for the VM this BMC boots.
    pub mem_mb: u32,
    /// vCPU count for the VM this BMC boots.
    pub cores: u32,
    /// `ComputerSystem.Name`.
    pub system_name: String,
    /// `ComputerSystem.Manufacturer`.
    pub manufacturer: String,
    /// `ComputerSystem.Model`.
    pub model: String,
    /// `ComputerSystem.UUID` / `ServiceRoot.UUID`.
    pub uuid: String,
    /// Extra TLS SANs beyond `localhost` / `127.0.0.1` — set when the BMC is reached
    /// by a name other than loopback.
    pub extra_sans: Vec<String>,
}

impl NodeConfig {
    /// A node with `system_id`, bound to an ephemeral loopback port, `admin` with no
    /// password, no local disk, 1024 MiB / 2 cores.
    pub fn new(system_id: impl Into<String>) -> Self {
        Self {
            system_id: system_id.into(),
            media_slot: crate::redfish::DEFAULT_MEDIA_ID.to_string(),
            username: "admin".into(),
            password: String::new(),
            bind: SocketAddr::from(([127, 0, 0, 1], 0)),
            local_disk: None,
            mem_mb: 1024,
            cores: 2,
            system_name: "draupnir-node".into(),
            manufacturer: "nordisk".into(),
            model: "draupnir Redfish/KVM".into(),
            uuid: "1f0d3a26-4c5b-4e77-9a2c-6b1e0d7a55f1".into(),
            extra_sans: Vec::new(),
        }
    }

    /// Set the Redfish account (builder style).
    pub fn credentials(mut self, username: impl Into<String>, password: impl Into<String>) -> Self {
        self.username = username.into();
        self.password = password.into();
        self
    }

    /// Set the address to bind (builder style).
    pub fn bind(mut self, bind: SocketAddr) -> Self {
        self.bind = bind;
        self
    }

    /// Set the `VirtualMedia` slot id (builder style).
    pub fn media_slot(mut self, slot: impl Into<String>) -> Self {
        self.media_slot = slot.into();
        self
    }

    /// Give the node a local disk — what an `Hdd` boot override boots (builder style).
    pub fn local_disk(mut self, disk: impl Into<String>) -> Self {
        self.local_disk = Some(disk.into());
        self
    }

    /// Size the VM this BMC boots (builder style).
    pub fn sized(mut self, mem_mb: u32, cores: u32) -> Self {
        self.mem_mb = mem_mb;
        self.cores = cores;
        self
    }

    /// Add TLS SANs (builder style).
    pub fn extra_sans<I, S>(mut self, sans: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.extra_sans.extend(sans.into_iter().map(Into::into));
        self
    }
}

// ===========================================================================
// The node's state
// ===========================================================================

/// The mutable state of the emulated node — everything the six endpoints read and
/// write. One `Mutex` guards it all: this is a control plane answering a handful of
/// requests, not a data path, and a single lock is the shape that cannot deadlock.
#[derive(Debug, Default)]
struct NodeState {
    /// The media URI exactly as the client named it (`VirtualMedia.Image`).
    image_uri: Option<String>,
    /// The local file that URI resolved to — what a boot actually attaches.
    image_path: Option<String>,
    /// `VirtualMedia.Inserted`.
    inserted: bool,
    /// `VirtualMedia.WriteProtected`.
    write_protected: bool,
    /// `Boot.BootSourceOverrideEnabled`.
    override_enabled: String,
    /// `Boot.BootSourceOverrideTarget`, once mapped.
    override_target: Option<BootTarget>,
    /// The machine the last successful power-on produced.
    machine: Option<Machine>,
    /// The [`BootSpec`] the last power-on actually handed the backend — the applied
    /// output, readable by a caller that wants to prove *what* was booted rather than
    /// that something was.
    last_spec: Option<BootSpec>,
    /// The last `ResetType` accepted.
    last_reset: Option<String>,
}

impl NodeState {
    fn fresh() -> Self {
        Self {
            override_enabled: wire::OVERRIDE_DISABLED.to_string(),
            ..Default::default()
        }
    }
}

// ===========================================================================
// The backend seam
// ===========================================================================

/// What this BMC can drive: anything that can be [`Boot`]ed and has a
/// [`Lifecycle`].
///
/// [`KvmBoot`] is the real one. The seam is a trait so the conformance suite can
/// front a **recording** backend that launches nothing — the wire shapes must be
/// provable on a box with no `/dev/kvm`, and a conformance test that needed a real
/// VM would be a conformance test nobody ran.
pub trait NodeBackend: Boot + Lifecycle + Send + Sync {}
impl<T: Boot + Lifecycle + Send + Sync> NodeBackend for T {}

// ===========================================================================
// The server
// ===========================================================================

/// A running Redfish service in front of a KVM host.
///
/// Started with [`start`](RedfishKvmServer::start) (real KVM) or
/// [`start_with`](RedfishKvmServer::start_with) (any [`NodeBackend`]). It listens on
/// its own background thread — a [`gatling::background::Job`], the constellation's
/// sanctioned home for a raw thread — until [`shutdown`](RedfishKvmServer::shutdown)
/// or drop.
pub struct RedfishKvmServer {
    cfg: NodeConfig,
    addr: SocketAddr,
    cert_pem: String,
    state: Arc<Mutex<NodeState>>,
    stop: Arc<AtomicBool>,
    /// The accept loop. `Option` so `shutdown` can take and join it.
    job: Option<gatling::background::Job<()>>,
    /// Kept typed (not just as a `NodeBackend`) so the serial console of the guest
    /// this BMC booted is readable — the applied output a caller needs to prove the
    /// override was honoured, rather than merely acknowledged.
    kvm: Option<Arc<KvmBoot>>,
}

impl std::fmt::Debug for RedfishKvmServer {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RedfishKvmServer")
            .field("addr", &self.addr)
            .field("system_id", &self.cfg.system_id)
            .field("media_slot", &self.cfg.media_slot)
            // Never the password, and never the private key (which is not held here
            // at all — only the public cert PEM).
            .field("cert_pem", &format_args!("<{} bytes>", self.cert_pem.len()))
            .finish_non_exhaustive()
    }
}

impl RedfishKvmServer {
    /// Start a BMC in front of **this box's KVM** ([`KvmBoot`]).
    pub fn start(cfg: NodeConfig) -> Result<Self> {
        let kvm = Arc::new(KvmBoot::new());
        let backend: Arc<dyn NodeBackend> = kvm.clone();
        let mut me = Self::start_with(cfg, backend)?;
        me.kvm = Some(kvm);
        Ok(me)
    }

    /// Start a BMC in front of an arbitrary [`NodeBackend`] — the seam the
    /// conformance suite uses to prove the wire with no VM anywhere.
    pub fn start_with(cfg: NodeConfig, backend: Arc<dyn NodeBackend>) -> Result<Self> {
        let (tls_config, cert_pem) = tls_identity(&cfg)?;
        let listener = TcpListener::bind(cfg.bind)
            .map_err(|e| Error::Backend(format!("redfish server bind {}: {e}", cfg.bind)))?;
        let addr = listener
            .local_addr()
            .map_err(|e| Error::Backend(format!("redfish server local_addr: {e}")))?;
        // Non-blocking accept + a short poll is what lets `shutdown` be prompt
        // without a self-connect trick; the accepted socket is put back into
        // blocking mode before the TLS handshake, which needs it.
        listener
            .set_nonblocking(true)
            .map_err(|e| Error::Backend(format!("redfish server set_nonblocking: {e}")))?;

        let state = Arc::new(Mutex::new(NodeState::fresh()));
        let stop = Arc::new(AtomicBool::new(false));

        let service = Service {
            routes: Routes::for_node(&cfg),
            cfg: cfg.clone(),
            state: state.clone(),
            backend,
        };
        let loop_stop = stop.clone();
        let job = gatling::background::Job::spawn(move || {
            accept_loop(listener, tls_config, service, loop_stop)
        });

        crate::functional_status(
            "draupnir/redfish-server",
            "start",
            true,
            &format!(
                "BMC for system `{}` listening on https://{addr} (slot `{}`)",
                cfg.system_id, cfg.media_slot
            ),
        );

        Ok(Self {
            cfg,
            addr,
            cert_pem,
            state,
            stop,
            job: Some(job),
            kvm: None,
        })
    }

    /// The address it is listening on (the real port when `bind` asked for `:0`).
    pub fn addr(&self) -> SocketAddr {
        self.addr
    }

    /// The service base URL, e.g. `https://127.0.0.1:41234`.
    pub fn base_url(&self) -> String {
        format!("https://{}", self.addr)
    }

    /// The **PEM certificate to pin** — feed it to
    /// [`RedfishBoot::pin_cert_pem`](crate::redfish::RedfishBoot::pin_cert_pem) so the
    /// client keeps full TLS verification on with this cert as its sole trusted root.
    pub fn cert_pem(&self) -> &str {
        &self.cert_pem
    }

    /// A [`BmcEndpoint`] addressing this server — what a spec's
    /// [`on_metal`](BootSpec::on_metal) takes.
    pub fn bmc_endpoint(&self) -> BmcEndpoint {
        BmcEndpoint {
            host: self.base_url(),
            username: self.cfg.username.clone(),
            system_id: self.cfg.system_id.clone(),
        }
    }

    /// The machine the last accepted power-on produced, if any.
    pub fn machine(&self) -> Option<Machine> {
        self.state.lock().unwrap().machine.clone()
    }

    /// **The [`BootSpec`] the BMC actually handed its backend** on the last power-on.
    ///
    /// This is the applied output at the spec level: it says *what* the node was told
    /// to boot, so a test can prove a boot override was honoured rather than merely
    /// acknowledged with a `204`.
    pub fn last_boot_spec(&self) -> Option<BootSpec> {
        self.state.lock().unwrap().last_spec.clone()
    }

    /// The media URI currently inserted (`VirtualMedia.Image`), if any.
    pub fn inserted_image(&self) -> Option<String> {
        let s = self.state.lock().unwrap();
        s.inserted.then(|| s.image_uri.clone()).flatten()
    }

    /// **The serial console of the guest this BMC booted** — the applied output at
    /// the machine level. `None` unless the server was started with
    /// [`start`](RedfishKvmServer::start) (a real [`KvmBoot`]) and a VM is live.
    ///
    /// Redfish itself has no "give me the console" GET, so this is deliberately a
    /// *host-side* accessor rather than an invented endpoint: an emulator that grew a
    /// non-standard resource would be a worse emulator.
    pub fn serial_log(&self) -> Option<String> {
        let kvm = self.kvm.as_ref()?;
        let machine = self.machine()?;
        kvm.serial_log(&machine)
    }

    /// Wait for a named line on the booted guest's console, or a deadline — the same
    /// [`Seen`] observation shape [`KvmBoot::await_serial_marker`] returns, because it
    /// *is* that call. `Err` when this server never booted a KVM guest.
    pub fn await_serial_marker(
        &self,
        marker: &str,
        budget: Duration,
        poll: Duration,
    ) -> Result<Seen> {
        let kvm = self.kvm.as_ref().ok_or_else(|| {
            Error::Backend(
                "this Redfish server does not front a KVM backend (started with start_with)".into(),
            )
        })?;
        let machine = self
            .machine()
            .ok_or_else(|| Error::Backend("no machine has been powered on yet".into()))?;
        kvm.await_serial_marker(&machine, marker, budget, poll)
    }

    /// Stop the listener and join its thread.
    pub fn shutdown(&mut self) {
        self.stop.store(true, Ordering::SeqCst);
        if let Some(job) = self.job.take() {
            let _ = job.join();
        }
    }
}

impl Drop for RedfishKvmServer {
    fn drop(&mut self) {
        self.shutdown();
        // A BMC that "powered off" by exiting would leave a QEMU process on the box.
        if let (Some(kvm), Some(machine)) = (self.kvm.as_ref(), self.machine()) {
            let _ = kvm.power_off(&machine);
        }
    }
}

// ===========================================================================
// TLS identity
// ===========================================================================

/// Mint the server's TLS identity and build a `rustls` config from it.
///
/// A genuine self-signed X.509 with `localhost`, `127.0.0.1`, the bound IP and any
/// [`NodeConfig::extra_sans`] as SANs. The private key never leaves this function's
/// return value — it goes straight into the `rustls` config and is never serialized
/// to disk, so no key material can be left behind or committed.
#[allow(clippy::type_complexity)]
fn tls_identity(cfg: &NodeConfig) -> Result<(Arc<rustls::ServerConfig>, String)> {
    let mut sans: Vec<String> = vec![
        "localhost".into(),
        "127.0.0.1".into(),
        "::1".into(),
        cfg.bind.ip().to_string(),
    ];
    sans.extend(cfg.extra_sans.iter().cloned());
    sans.sort();
    sans.dedup();

    let mut params = rcgen::CertificateParams::new(sans)
        .map_err(|e| Error::Backend(format!("redfish server cert params: {e}")))?;
    params.distinguished_name = {
        let mut dn = rcgen::DistinguishedName::new();
        dn.push(
            rcgen::DnType::CommonName,
            format!("draupnir BMC {}", cfg.system_id),
        );
        dn.push(rcgen::DnType::OrganizationName, "nordisk");
        dn
    };
    let key = rcgen::KeyPair::generate()
        .map_err(|e| Error::Backend(format!("redfish server keypair: {e}")))?;
    let cert = params
        .self_signed(&key)
        .map_err(|e| Error::Backend(format!("redfish server self-sign: {e}")))?;
    let cert_pem = cert.pem();

    let cert_der = rustls::pki_types::CertificateDer::from(cert.der().to_vec());
    let key_der = rustls::pki_types::PrivateKeyDer::try_from(key.serialize_der())
        .map_err(|e| Error::Backend(format!("redfish server key DER: {e}")))?;

    let provider = Arc::new(rustls::crypto::ring::default_provider());
    let tls = rustls::ServerConfig::builder_with_provider(provider)
        .with_safe_default_protocol_versions()
        .map_err(|e| Error::Backend(format!("redfish server tls versions: {e}")))?
        .with_no_client_auth()
        .with_single_cert(vec![cert_der], key_der)
        .map_err(|e| Error::Backend(format!("redfish server tls cert: {e}")))?;

    Ok((Arc::new(tls), cert_pem))
}

// ===========================================================================
// The accept loop
// ===========================================================================

/// How long a single connection may take to deliver its request / read its response.
/// A BMC exchange is a few hundred bytes; a peer that stalls longer than this is
/// wedged, and holding the (deliberately serial) control plane for it would be worse
/// than dropping it.
const CONNECTION_TIMEOUT: Duration = Duration::from_secs(30);

/// The idle poll between `accept` attempts — what makes shutdown prompt.
const ACCEPT_POLL: Duration = Duration::from_millis(5);

fn accept_loop(
    listener: TcpListener,
    tls: Arc<rustls::ServerConfig>,
    service: Service,
    stop: Arc<AtomicBool>,
) {
    while !stop.load(Ordering::SeqCst) {
        match listener.accept() {
            Ok((sock, _peer)) => {
                if sock.set_nonblocking(false).is_err() {
                    continue;
                }
                let _ = sock.set_read_timeout(Some(CONNECTION_TIMEOUT));
                let _ = sock.set_write_timeout(Some(CONNECTION_TIMEOUT));
                serve_connection(sock, &tls, &service);
            }
            Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                std::thread::sleep(ACCEPT_POLL);
            }
            Err(_) => break,
        }
    }
}

/// Serve **one** request on one TLS connection, then close it gracefully.
///
/// # Why one request per connection, and not keep-alive
///
/// This was keep-alive first, and it produced a real, reproducible flake: measured on
/// this box 2026-08-15, **3 failures in 28** full runs of `cargo test --features
/// redfish-server,backend-redfish --test redfish_server_conformance`, each an
/// `ECONNRESET` raised by the client. Instrumenting the accept loop showed the classic
/// HTTP/1.1 **idle-close race**: the server had answered two requests on a pooled
/// connection, the client's pool retired it, and the client's next request went out on
/// the socket the server was already closing. After this change: **0 failures in 40**
/// consecutive runs.
///
/// The race is inherent to keep-alive and every client is supposed to retry through
/// it. A flaky proof is worse than a slow one, so this service does the thing a
/// server is always allowed to do: answer one request, say `Connection: close`, and
/// shut the connection down cleanly. Plenty of real BMCs behave exactly this way. The
/// cost is a TLS handshake per request — about a millisecond on loopback, for a
/// control plane that makes single-digit numbers of calls per boot.
///
/// It also removes a second defect keep-alive had: the accept loop is deliberately
/// **serial** (a BMC is not a data path), so a connection parked in `read` waiting for
/// a follow-up request that never came would hold the whole service — and would make
/// [`RedfishKvmServer::shutdown`] block for the full [`CONNECTION_TIMEOUT`].
fn serve_connection(sock: TcpStream, tls: &Arc<rustls::ServerConfig>, service: &Service) {
    let Ok(conn) = rustls::ServerConnection::new(tls.clone()) else {
        return;
    };
    let mut stream = rustls::StreamOwned::new(conn, sock);
    let mut buf: Vec<u8> = Vec::with_capacity(2048);
    match read_request(&mut stream, &mut buf) {
        // A clean close, or a peer that went away mid-request: nothing to answer.
        Ok(None) => {}
        Ok(Some(request)) => {
            let response = service
                .dispatch(&request)
                .with_header("Connection", "close");
            let _ = write_response(&mut stream, &response);
        }
        Err(resp) => {
            let _ = write_response(&mut stream, &resp.with_header("Connection", "close"));
        }
    }
    graceful_close(stream);
}

/// Close a TLS connection so the peer sees a clean shutdown rather than a reset.
///
/// Three steps, each load-bearing: send the TLS `close_notify` (so the client's TLS
/// layer sees an orderly end of stream, not a truncation), `shutdown(Write)` to send
/// the FIN, and then **drain** whatever the peer still had in flight. That last one is
/// the reset: the kernel emits RST instead of FIN when a socket is closed with unread
/// data in its receive queue, which is precisely how a well-behaved client ends up
/// reporting "connection reset by peer" against a server that did nothing wrong.
fn graceful_close(mut stream: rustls::StreamOwned<rustls::ServerConnection, TcpStream>) {
    stream.conn.send_close_notify();
    let _ = stream.flush();
    let _ = stream.sock.shutdown(std::net::Shutdown::Write);
    let _ = stream
        .sock
        .set_read_timeout(Some(Duration::from_millis(250)));
    let mut sink = [0u8; 1024];
    // Bounded: drain what is there, then stop. An endless drain would be a new way
    // for one peer to hold the serial accept loop.
    for _ in 0..16 {
        match stream.sock.read(&mut sink) {
            Ok(0) | Err(_) => break,
            Ok(_) => continue,
        }
    }
}

// ===========================================================================
// Minimal HTTP/1.1 — pure std, no framework
// ===========================================================================

/// The largest request head this service will read. A Redfish request head is a few
/// hundred bytes; anything past this is not a BMC client.
const MAX_HEAD: usize = 16 * 1024;
/// The largest request body. The biggest legitimate body here is an `InsertMedia`
/// with a long URI.
const MAX_BODY: usize = 1024 * 1024;

/// A parsed HTTP request.
struct Request {
    method: String,
    /// The percent-decoded path, without the query string.
    path: String,
    headers: BTreeMap<String, String>,
    body: Vec<u8>,
}

impl Request {
    fn header(&self, name: &str) -> Option<&str> {
        self.headers.get(&name.to_ascii_lowercase()).map(String::as_str)
    }
}

/// A response to write.
struct Response {
    status: u16,
    headers: Vec<(String, String)>,
    body: Vec<u8>,
}

impl Response {
    fn new(status: u16) -> Self {
        Self {
            status,
            headers: Vec::new(),
            body: Vec::new(),
        }
    }

    fn with_header(mut self, name: &str, value: impl Into<String>) -> Self {
        self.headers.push((name.to_string(), value.into()));
        self
    }

    /// A JSON resource body.
    fn json(status: u16, value: &Value) -> Self {
        let mut r = Self::new(status);
        r.body = serde_json::to_vec(value).unwrap_or_else(|_| b"{}".to_vec());
        r.headers.push((
            "Content-Type".into(),
            "application/json;charset=utf-8".into(),
        ));
        r
    }

    /// `204 No Content` — what a successful Redfish action or PATCH returns.
    fn no_content() -> Self {
        Self::new(204)
    }

    /// A DMTF-shaped `error` payload (`redfish-error.v1_0_2`): `error.code`,
    /// `error.message`, and one `@Message.ExtendedInfo` entry carrying the registry
    /// `MessageId`, the rendered `Message`, its `MessageArgs`, `MessageSeverity` and
    /// `Resolution`.
    ///
    /// `detail`, when given, is appended to the extended info's `Resolution` — so a
    /// refusal always **names the thing it refused** even where DMTF's registry
    /// template has no argument for it.
    fn error(msg: Message, args: &[&str], detail: Option<&str>) -> Self {
        let rendered = msg.render(args);
        let resolution = match detail {
            Some(d) => format!("{} {d}", msg.resolution),
            None => msg.resolution.to_string(),
        };
        let mut info = json!({
            "@odata.type": "#Message.v1_1_2.Message",
            "MessageId": msg.message_id(),
            "Message": rendered,
            "MessageSeverity": msg.severity,
            "Resolution": resolution,
        });
        if !args.is_empty() {
            info["MessageArgs"] = json!(args);
        }
        Response::json(
            msg.status,
            &json!({
                "error": {
                    "code": msg.message_id(),
                    "message": rendered,
                    "@Message.ExtendedInfo": [info],
                }
            }),
        )
    }

    /// `405 Method Not Allowed` with the mandatory `Allow` header — DSP0266 requires
    /// the header, and a client that reads it can discover the right verb.
    fn method_not_allowed(allow: &str) -> Self {
        Response::error(
            ACTION_NOT_SUPPORTED,
            &["the requested method"],
            Some(&format!("Allowed methods: {allow}.")),
        )
        .with_status(405)
        .with_header("Allow", allow)
    }

    fn with_status(mut self, status: u16) -> Self {
        self.status = status;
        self
    }
}

fn reason(status: u16) -> &'static str {
    match status {
        200 => "OK",
        201 => "Created",
        204 => "No Content",
        400 => "Bad Request",
        401 => "Unauthorized",
        404 => "Not Found",
        405 => "Method Not Allowed",
        413 => "Payload Too Large",
        500 => "Internal Server Error",
        501 => "Not Implemented",
        _ => "Status",
    }
}

/// Read one request off `stream`, buffering leftovers for the next keep-alive round.
///
/// `Ok(None)` is a clean close. `Err(response)` is a request this service will not
/// even parse (too large, unreadable) and carries what to answer with.
fn read_request<S: Read>(stream: &mut S, buf: &mut Vec<u8>) -> std::result::Result<Option<Request>, Response> {
    // 1. Head — read until CRLFCRLF.
    let head_end = loop {
        if let Some(p) = find_subsequence(buf, b"\r\n\r\n") {
            break p;
        }
        if buf.len() > MAX_HEAD {
            return Err(Response::error(
                MALFORMED_JSON,
                &[],
                Some("The request head exceeded this service's limit."),
            )
            .with_status(413));
        }
        let mut chunk = [0u8; 1024];
        match stream.read(&mut chunk) {
            Ok(0) => return Ok(None),
            Ok(n) => buf.extend_from_slice(&chunk[..n]),
            Err(_) => return Ok(None),
        }
    };

    let head = String::from_utf8_lossy(&buf[..head_end]).into_owned();
    let mut lines = head.split("\r\n");
    let request_line = lines.next().unwrap_or_default();
    let mut parts = request_line.split_whitespace();
    let method = parts.next().unwrap_or_default().to_string();
    let raw_target = parts.next().unwrap_or_default();
    if method.is_empty() || raw_target.is_empty() {
        return Ok(None);
    }
    let path_part = raw_target.split('?').next().unwrap_or(raw_target);
    let path = percent_decode(path_part);

    let mut headers = BTreeMap::new();
    for line in lines {
        if let Some((k, v)) = line.split_once(':') {
            headers.insert(k.trim().to_ascii_lowercase(), v.trim().to_string());
        }
    }

    // 2. Body — Content-Length only. Redfish clients do not chunk their action
    //    bodies, and a service that guessed at a chunked body would be guessing.
    let len: usize = headers
        .get("content-length")
        .and_then(|v| v.parse().ok())
        .unwrap_or(0);
    if len > MAX_BODY {
        return Err(Response::error(
            MALFORMED_JSON,
            &[],
            Some("The request body exceeded this service's limit."),
        )
        .with_status(413));
    }
    let body_start = head_end + 4;
    while buf.len() < body_start + len {
        let mut chunk = [0u8; 4096];
        match stream.read(&mut chunk) {
            Ok(0) => return Ok(None),
            Ok(n) => buf.extend_from_slice(&chunk[..n]),
            Err(_) => return Ok(None),
        }
    }
    let body = buf[body_start..body_start + len].to_vec();
    // Keep whatever the peer pipelined behind this request.
    buf.drain(..body_start + len);

    Ok(Some(Request {
        method,
        path,
        headers,
        body,
    }))
}

fn write_response<S: Write>(stream: &mut S, resp: &Response) -> std::io::Result<()> {
    let mut out = Vec::with_capacity(256 + resp.body.len());
    out.extend_from_slice(
        format!("HTTP/1.1 {} {}\r\n", resp.status, reason(resp.status)).as_bytes(),
    );
    // Redfish requires OData-Version on every response.
    out.extend_from_slice(b"OData-Version: 4.0\r\n");
    out.extend_from_slice(b"Cache-Control: no-cache\r\n");
    for (k, v) in &resp.headers {
        out.extend_from_slice(format!("{k}: {v}\r\n").as_bytes());
    }
    out.extend_from_slice(format!("Content-Length: {}\r\n\r\n", resp.body.len()).as_bytes());
    out.extend_from_slice(&resp.body);
    stream.write_all(&out)?;
    stream.flush()
}

fn find_subsequence(haystack: &[u8], needle: &[u8]) -> Option<usize> {
    haystack
        .windows(needle.len())
        .position(|w| w == needle)
}

/// Percent-decode a URL path segment sequence. Invalid escapes are left verbatim
/// rather than dropped, so a path this service does not recognise fails as
/// "not found at *that* URI" and names the URI the client actually sent.
fn percent_decode(s: &str) -> String {
    let bytes = s.as_bytes();
    let mut out = Vec::with_capacity(bytes.len());
    let mut i = 0;
    while i < bytes.len() {
        if bytes[i] == b'%' && i + 2 < bytes.len() {
            let hex = std::str::from_utf8(&bytes[i + 1..i + 3]).ok();
            if let Some(b) = hex.and_then(|h| u8::from_str_radix(h, 16).ok()) {
                out.push(b);
                i += 3;
                continue;
            }
        }
        out.push(bytes[i]);
        i += 1;
    }
    String::from_utf8_lossy(&out).into_owned()
}

// ===========================================================================
// Routing
// ===========================================================================

/// The service's routing table — every path built from [`crate::redfish::wire`], the
/// same module the CLIENT builds its URLs from. This is the whole of the anti-mirror
/// discipline at the URL level: there is no second spelling to drift from.
#[derive(Debug, Clone)]
struct Routes {
    system: String,
    reset: String,
    vm_collection: String,
    vm: String,
    vm_insert: String,
    vm_eject: String,
}

impl Routes {
    fn for_node(cfg: &NodeConfig) -> Self {
        let id = &cfg.system_id;
        let slot = &cfg.media_slot;
        Self {
            system: wire::system_path(id),
            reset: wire::reset_path(id),
            vm_collection: wire::virtual_media_collection_path(id),
            vm: wire::virtual_media_path(id, slot),
            vm_insert: wire::virtual_media_action_path(id, slot, wire::INSERT_MEDIA),
            vm_eject: wire::virtual_media_action_path(id, slot, wire::EJECT_MEDIA),
        }
    }
}

/// Everything a request needs: the routes, the node config, its state, its backend.
struct Service {
    routes: Routes,
    cfg: NodeConfig,
    state: Arc<Mutex<NodeState>>,
    backend: Arc<dyn NodeBackend>,
}

impl Service {
    fn dispatch(&self, req: &Request) -> Response {
        let path = req.path.as_str();
        let m = req.method.as_str();

        // The two unauthenticated URIs DSP0266 allows: the protocol-version probe
        // and the service root.
        if path == wire::PROTOCOL_VERSION_PATH {
            return match m {
                "GET" | "HEAD" => Response::json(200, &json!({ "v1": wire::SERVICE_ROOT_PATH })),
                _ => Response::method_not_allowed("GET"),
            };
        }
        if path == wire::SERVICE_ROOT_PATH || path == wire::SERVICE_ROOT_PATH_BARE {
            return match m {
                "GET" | "HEAD" => Response::json(200, &self.service_root()),
                _ => Response::method_not_allowed("GET"),
            };
        }

        if let Some(unauthorized) = self.check_auth(req) {
            return unauthorized;
        }

        let r = &self.routes;
        if path == wire::SYSTEMS_PATH {
            return match m {
                "GET" | "HEAD" => Response::json(200, &self.system_collection()),
                _ => Response::method_not_allowed("GET"),
            };
        }
        if path == wire::SESSIONS_PATH {
            return match m {
                "GET" | "HEAD" => Response::json(200, &self.session_collection()),
                _ => Response::method_not_allowed("GET"),
            };
        }
        if path == r.system {
            return match m {
                "GET" | "HEAD" => Response::json(200, &self.computer_system()),
                "PATCH" => self.patch_system(req),
                _ => Response::method_not_allowed("GET, PATCH"),
            };
        }
        if path == r.reset {
            return match m {
                "POST" => self.post_reset(req),
                _ => Response::method_not_allowed("POST"),
            };
        }
        if path == r.vm_collection {
            return match m {
                "GET" | "HEAD" => Response::json(200, &self.virtual_media_collection()),
                _ => Response::method_not_allowed("GET"),
            };
        }
        if path == r.vm {
            return match m {
                "GET" | "HEAD" => Response::json(200, &self.virtual_media()),
                _ => Response::method_not_allowed("GET"),
            };
        }
        if path == r.vm_insert {
            return match m {
                "POST" => self.post_insert_media(req),
                _ => Response::method_not_allowed("POST"),
            };
        }
        if path == r.vm_eject {
            return match m {
                "POST" => self.post_eject_media(),
                _ => Response::method_not_allowed("POST"),
            };
        }

        Response::error(RESOURCE_NOT_FOUND, &["Resource", path], None)
    }

    /// `None` when the request is authorized. Everything below the service root
    /// needs a Basic credential matching the configured account.
    fn check_auth(&self, req: &Request) -> Option<Response> {
        let ok = req
            .header("authorization")
            .and_then(wire::parse_basic_auth)
            .is_some_and(|(u, p)| u == self.cfg.username && p == self.cfg.password);
        if ok {
            return None;
        }
        Some(
            Response::error(NO_VALID_SESSION, &[], None)
                .with_header("WWW-Authenticate", "Basic realm=\"RedfishService\""),
        )
    }

    // ── resources ───────────────────────────────────────────────────────────

    fn service_root(&self) -> Value {
        json!({
            "@odata.id": wire::SERVICE_ROOT_PATH,
            "@odata.type": odata_type::SERVICE_ROOT,
            "Id": "RootService",
            "Name": "Root Service",
            "RedfishVersion": REDFISH_VERSION,
            "UUID": self.cfg.uuid,
            "Product": "draupnir",
            "Vendor": "nordisk",
            "Systems": { "@odata.id": wire::SYSTEMS_PATH },
            // REQUIRED by the ServiceRoot schema, and it must resolve — hence the
            // (read-only, empty) SessionCollection this service also answers.
            "Links": { "Sessions": { "@odata.id": wire::SESSIONS_PATH } },
        })
    }

    fn system_collection(&self) -> Value {
        json!({
            "@odata.id": wire::SYSTEMS_PATH,
            "@odata.type": odata_type::SYSTEM_COLLECTION,
            "Name": "Computer System Collection",
            "Members": [ { "@odata.id": self.routes.system } ],
            "Members@odata.count": 1,
        })
    }

    /// An EMPTY session collection.
    ///
    /// It exists because `ServiceRoot.Links.Sessions` is a **required** property whose
    /// pointer must resolve; it is empty because this service authenticates with HTTP
    /// Basic only. Session *creation* is a named gap, not a silent one: a `POST` here
    /// is answered `405` with `Allow: GET`, so a client that needs session auth is
    /// told so instead of being handed a plausible-looking token.
    fn session_collection(&self) -> Value {
        json!({
            "@odata.id": wire::SESSIONS_PATH,
            "@odata.type": odata_type::SESSION_COLLECTION,
            "Name": "Session Collection",
            "Members": [],
            "Members@odata.count": 0,
        })
    }

    fn computer_system(&self) -> Value {
        let s = self.state.lock().unwrap();
        let power = self.observe_power(&s);
        let target = s
            .override_target
            .map_or(wire::BOOT_TARGET_NONE, wire::target_str);
        json!({
            "@odata.id": self.routes.system,
            "@odata.type": odata_type::SYSTEM,
            "Id": self.cfg.system_id,
            "Name": self.cfg.system_name,
            "SystemType": "Physical",
            "Manufacturer": self.cfg.manufacturer,
            "Model": self.cfg.model,
            "UUID": self.cfg.uuid,
            "PowerState": power_state_str(power),
            "Status": { "State": "Enabled", "Health": "OK" },
            "Boot": {
                prop::BOOT_SOURCE_OVERRIDE_ENABLED: s.override_enabled,
                prop::BOOT_SOURCE_OVERRIDE_TARGET: target,
                prop::BOOT_SOURCE_OVERRIDE_MODE: "UEFI",
                wire::allowable_values_key(prop::BOOT_SOURCE_OVERRIDE_TARGET):
                    wire::BOOT_TARGET_ALLOWABLE,
            },
            "MemorySummary": { "TotalSystemMemoryGiB": self.cfg.mem_mb as f64 / 1024.0 },
            "ProcessorSummary": { "Count": self.cfg.cores },
            "VirtualMedia": { "@odata.id": self.routes.vm_collection },
            "Actions": {
                wire::ACTION_RESET: {
                    prop::TARGET: self.routes.reset,
                    wire::allowable_values_key(prop::RESET_TYPE): wire::RESET_TYPE_ALLOWABLE,
                }
            },
        })
    }

    fn virtual_media_collection(&self) -> Value {
        json!({
            "@odata.id": self.routes.vm_collection,
            "@odata.type": odata_type::VIRTUAL_MEDIA_COLLECTION,
            "Name": "Virtual Media Services",
            "Members": [ { "@odata.id": self.routes.vm } ],
            "Members@odata.count": 1,
        })
    }

    fn virtual_media(&self) -> Value {
        let s = self.state.lock().unwrap();
        json!({
            "@odata.id": self.routes.vm,
            "@odata.type": odata_type::VIRTUAL_MEDIA,
            "Id": self.cfg.media_slot,
            "Name": "Virtual CD",
            "MediaTypes": ["CD", "DVD"],
            "ConnectedVia": if s.inserted { "URI" } else { "NotConnected" },
            prop::IMAGE: s.image_uri,
            prop::IMAGE_NAME: s.image_uri.as_deref().and_then(basename),
            prop::INSERTED: s.inserted,
            prop::WRITE_PROTECTED: s.write_protected,
            "Actions": {
                wire::ACTION_INSERT_MEDIA: { prop::TARGET: self.routes.vm_insert },
                wire::ACTION_EJECT_MEDIA:  { prop::TARGET: self.routes.vm_eject },
            },
        })
    }

    // ── actions ─────────────────────────────────────────────────────────────

    fn body_json(&self, req: &Request) -> std::result::Result<Value, Response> {
        if req.body.is_empty() {
            return Ok(json!({}));
        }
        serde_json::from_slice(&req.body)
            .map_err(|e| Response::error(MALFORMED_JSON, &[], Some(&format!("Parser said: {e}."))))
    }

    /// `POST .../VirtualMedia/{slot}/Actions/VirtualMedia.InsertMedia`
    fn post_insert_media(&self, req: &Request) -> Response {
        let body = match self.body_json(req) {
            Ok(b) => b,
            Err(r) => return r,
        };
        let insert = match wire::read_insert_media_body(&body) {
            Ok(i) => i,
            Err(missing) => {
                return Response::error(
                    ACTION_PARAMETER_MISSING,
                    &[wire::ACTION_INSERT_MEDIA, missing],
                    None,
                )
            }
        };
        // RED CONTROL: a media URI that resolves to nothing on this host is refused
        // BY NAME, before anything is mounted. An emulator that accepted it would
        // hand a later boot an empty tray and report the failure as "the appliance
        // did not come up" — a different bug with a different owner.
        let path = match resolve_media(&insert.image) {
            Ok(p) => p,
            Err(why) => {
                return Response::error(
                    RESOURCE_MISSING_AT_URI,
                    &[&insert.image],
                    Some(&why),
                )
            }
        };
        let mut s = self.state.lock().unwrap();
        s.image_uri = Some(insert.image.clone());
        s.image_path = Some(path.clone());
        s.inserted = insert.inserted;
        s.write_protected = insert.write_protected;
        drop(s);
        crate::functional_status(
            "draupnir/redfish-server",
            "InsertMedia",
            true,
            &format!("`{}` mounted from `{path}`", insert.image),
        );
        Response::no_content()
    }

    /// `POST .../VirtualMedia/{slot}/Actions/VirtualMedia.EjectMedia`
    fn post_eject_media(&self) -> Response {
        let mut s = self.state.lock().unwrap();
        s.image_uri = None;
        s.image_path = None;
        s.inserted = false;
        s.write_protected = false;
        Response::no_content()
    }

    /// `PATCH /redfish/v1/Systems/{id}` — the one-time boot override.
    fn patch_system(&self, req: &Request) -> Response {
        let body = match self.body_json(req) {
            Ok(b) => b,
            Err(r) => return r,
        };
        let Some(ovr) = wire::read_boot_override_body(&body) else {
            // A PATCH that names nothing this service owns changes nothing. It is not
            // an error (Redfish PATCH is a partial update), and it must not be
            // reported as one.
            return Response::no_content();
        };

        let mut s = self.state.lock().unwrap();
        if let Some(enabled) = &ovr.enabled {
            if !wire::OVERRIDE_ENABLED_ALLOWABLE.contains(&enabled.as_str()) {
                return Response::error(
                    PROPERTY_VALUE_NOT_IN_LIST,
                    &[enabled, prop::BOOT_SOURCE_OVERRIDE_ENABLED],
                    Some(&format!(
                        "Acceptable values: {}.",
                        wire::OVERRIDE_ENABLED_ALLOWABLE.join(", ")
                    )),
                );
            }
            s.override_enabled = enabled.clone();
        }
        if let Some(raw) = &ovr.target {
            if raw == wire::BOOT_TARGET_NONE {
                s.override_target = None;
            } else {
                // A target this node cannot actually honour is REFUSED, never quietly
                // coerced into one it can. `Pxe`/`BiosSetup` are legal DMTF tokens
                // that this KVM front end has nothing to do with, and answering 204
                // for them would be the exact "accepted but not applied" lie.
                match wire::target_from_str(raw)
                    .and_then(|t| BootOrder::from_redfish_target(Some(t)).map(|_| t))
                {
                    Some(t) => s.override_target = Some(t),
                    None => {
                        return Response::error(
                            PROPERTY_VALUE_NOT_IN_LIST,
                            &[raw, prop::BOOT_SOURCE_OVERRIDE_TARGET],
                            Some(&format!(
                                "This node honours: {}.",
                                boot_targets_this_node_honours().join(", ")
                            )),
                        )
                    }
                }
            }
        }
        let applied = format!(
            "{}/{}",
            s.override_enabled,
            s.override_target.map_or(wire::BOOT_TARGET_NONE, wire::target_str)
        );
        drop(s);
        crate::functional_status(
            "draupnir/redfish-server",
            "BootOverride",
            true,
            &format!("boot override set to {applied}"),
        );
        Response::no_content()
    }

    /// `POST /redfish/v1/Systems/{id}/Actions/ComputerSystem.Reset`
    fn post_reset(&self, req: &Request) -> Response {
        let body = match self.body_json(req) {
            Ok(b) => b,
            Err(r) => return r,
        };
        let reset_type = match wire::read_reset_body(&body) {
            Ok(t) => t,
            Err(missing) => {
                return Response::error(
                    ACTION_PARAMETER_MISSING,
                    &[wire::ACTION_RESET, missing],
                    None,
                )
            }
        };
        // RED CONTROL: a ResetType outside the advertised list is refused BY NAME,
        // quoting the value, the parameter and the action — DMTF's own
        // ActionParameterValueNotInList.
        if !wire::is_allowable_reset_type(&reset_type) {
            return Response::error(
                ACTION_PARAMETER_VALUE_NOT_IN_LIST,
                &[&reset_type, prop::RESET_TYPE, wire::ACTION_RESET],
                Some(&format!(
                    "Acceptable values: {}.",
                    wire::RESET_TYPE_ALLOWABLE.join(", ")
                )),
            );
        }

        match reset_type.as_str() {
            wire::RESET_ON | wire::RESET_FORCE_ON => self.power_up(&reset_type),
            wire::RESET_FORCE_OFF | wire::RESET_GRACEFUL_SHUTDOWN => self.power_down(&reset_type),
            wire::RESET_FORCE_RESTART | wire::RESET_GRACEFUL_RESTART => {
                self.power_down(&reset_type);
                self.power_up(&reset_type)
            }
            // Unreachable while RESET_TYPE_ALLOWABLE and this match agree; if they
            // ever stop agreeing, say so instead of pretending to act.
            other => Response::error(
                ACTION_PARAMETER_VALUE_NOT_IN_LIST,
                &[other, prop::RESET_TYPE, wire::ACTION_RESET],
                Some("This service advertises the value but does not implement it."),
            ),
        }
    }

    /// Power the node on: build the [`BootSpec`] the current state describes and hand
    /// it to the backend. **This is where the override becomes applied output** — the
    /// spec that leaves here is the machine that gets booted.
    fn power_up(&self, reset_type: &str) -> Response {
        let spec = {
            let s = self.state.lock().unwrap();
            match self.spec_for(&s) {
                Ok(spec) => spec,
                Err(why) => {
                    crate::functional_status(
                        "draupnir/redfish-server",
                        "Reset",
                        false,
                        &format!("{reset_type} refused: {why}"),
                    );
                    return Response::error(GENERAL_ERROR, &[], Some(&why));
                }
            }
        };
        match self.backend.boot(&spec) {
            Ok(machine) => {
                let mut s = self.state.lock().unwrap();
                s.machine = Some(machine);
                s.last_spec = Some(spec.clone());
                s.last_reset = Some(reset_type.to_string());
                // A ONE-TIME override is CONSUMED by the boot it applied to — DSP0266
                // says the service resets it to Disabled/None. Modelling that is what
                // stops a stale `Once` from silently steering a later boot.
                if s.override_enabled == wire::OVERRIDE_ONCE {
                    s.override_enabled = wire::OVERRIDE_DISABLED.to_string();
                    s.override_target = None;
                }
                drop(s);
                crate::functional_status(
                    "draupnir/redfish-server",
                    "Reset",
                    true,
                    &format!(
                        "{reset_type}: booted `{}` (order {:?}, medium {:?})",
                        spec.name,
                        spec.boot_order,
                        spec.medium_path()
                    ),
                );
                Response::no_content()
            }
            Err(e) => {
                crate::functional_status(
                    "draupnir/redfish-server",
                    "Reset",
                    false,
                    &format!("{reset_type} failed: {e}"),
                );
                Response::error(GENERAL_ERROR, &[], Some(&format!("The node's backend refused to start: {e}")))
            }
        }
    }

    fn power_down(&self, reset_type: &str) -> Response {
        let machine = self.state.lock().unwrap().machine.clone();
        let Some(machine) = machine else {
            // Powering off an already-off node is a no-op, not a fault.
            return Response::no_content();
        };
        match self.backend.power_off(&machine) {
            Ok(()) => {
                let mut s = self.state.lock().unwrap();
                s.machine = None;
                s.last_reset = Some(reset_type.to_string());
                Response::no_content()
            }
            Err(e) => Response::error(
                GENERAL_ERROR,
                &[],
                Some(&format!("The node's backend refused to power off: {e}")),
            ),
        }
    }

    /// **The translation, in one function.** Redfish state → a draupnir [`BootSpec`].
    ///
    /// The boot order comes from [`BootOrder::from_redfish_target`] — the inverse of
    /// the mapping the CLIENT uses — and then each order selects the constructor that
    /// already expresses it:
    ///
    /// * [`BootOrder::Medium`] → [`BootSpec::iso_boot`] (the inserted media is the
    ///   whole payload; `-cdrom` + `-boot d`).
    /// * [`BootOrder::Disk`] → [`BootSpec::kvm_boot_installed_disk`], which by
    ///   construction carries **no medium at all** ([`BootSpec::validate`] refuses
    ///   `Disk` + a medium). That is what makes an `Hdd` override *applied* rather
    ///   than merely acknowledged: the ISO is not attached to the machine.
    /// * [`BootOrder::Auto`] → the node's own order: the medium if a tray is loaded,
    ///   else the local disk.
    ///
    /// Every refusal names what was missing, because "the appliance did not come up"
    /// and "you never inserted an ISO" are different failures with different owners.
    fn spec_for(&self, s: &NodeState) -> std::result::Result<BootSpec, String> {
        let effective_target = if s.override_enabled == wire::OVERRIDE_DISABLED {
            None
        } else {
            s.override_target
        };
        let order = BootOrder::from_redfish_target(effective_target).ok_or_else(|| {
            format!(
                "boot override target {:?} is not a boot order this node can apply",
                effective_target
            )
        })?;

        let name = format!("redfish-{}", self.cfg.system_id);
        let medium = s.inserted.then(|| s.image_path.clone()).flatten();

        let mut spec = match order {
            BootOrder::Medium => {
                let iso = medium.ok_or_else(|| {
                    format!(
                        "the boot override asks for {} but no virtual media is inserted in slot `{}`",
                        wire::target_str(BootTarget::Cd),
                        self.cfg.media_slot
                    )
                })?;
                BootSpec::iso_boot(name, iso)
            }
            BootOrder::Disk => {
                let disk = self.cfg.local_disk.clone().ok_or_else(|| {
                    format!(
                        "the boot override asks for {} but this node has no local disk configured \
                         (NodeConfig::local_disk)",
                        wire::target_str(BootTarget::Hdd)
                    )
                })?;
                BootSpec::kvm_boot_installed_disk(name, disk)
            }
            BootOrder::Auto => match (medium, self.cfg.local_disk.clone()) {
                (Some(iso), _) => BootSpec::iso_boot(name, iso),
                (None, Some(disk)) => BootSpec::kvm_boot_installed_disk(name, disk),
                (None, None) => {
                    return Err(
                        "this node has neither virtual media inserted nor a local disk to boot"
                            .into(),
                    )
                }
            },
        };
        spec.mem_mb = self.cfg.mem_mb;
        spec.cores = self.cfg.cores;
        // Validate here, so a spec this service could not have booted is refused with
        // draupnir's own reason rather than QEMU's, from inside the Redfish action.
        spec.validate().map_err(|e| e.to_string())?;
        Ok(spec)
    }

    /// The node's power state, read from the backend when there is a machine.
    fn observe_power(&self, s: &NodeState) -> PowerState {
        match &s.machine {
            None => PowerState::Off,
            Some(m) => self.backend.status(m).unwrap_or(PowerState::Unknown),
        }
    }
}

/// The `BootSourceOverrideTarget` tokens this node will actually apply — the subset
/// of [`wire::BOOT_TARGET_ALLOWABLE`] that maps onto a [`BootOrder`].
fn boot_targets_this_node_honours() -> Vec<&'static str> {
    wire::BOOT_TARGET_ALLOWABLE
        .iter()
        .copied()
        .filter(|t| {
            wire::target_from_str(t)
                .and_then(|bt| BootOrder::from_redfish_target(Some(bt)))
                .is_some()
        })
        .chain(std::iter::once(wire::BOOT_TARGET_NONE))
        .collect()
}

/// The Redfish `PowerState` token for a draupnir [`PowerState`].
fn power_state_str(p: PowerState) -> &'static str {
    match p {
        PowerState::On => "On",
        PowerState::Off => "Off",
        // DMTF has no "unknown" PowerState token; the honest report for a node whose
        // state cannot be observed is to omit the claim, and `Off` would be a claim.
        PowerState::Unknown => "Paused",
    }
}

/// The last path segment of a media URI, for `VirtualMedia.ImageName`.
fn basename(uri: &str) -> Option<String> {
    uri.rsplit(['/', '\\']).next().map(str::to_string).filter(|s| !s.is_empty())
}

/// **Resolve a Redfish media URI onto a local file.**
///
/// A real BMC *pulls* the image over http(s)/NFS/CIFS. This service fronts local
/// KVM, so it resolves a `file://` URI or an absolute path and **refuses a remote
/// scheme by name** rather than pretending to have fetched it. That limitation is a
/// property of the emulator, not of the protocol: draupnir's client is unchanged and
/// still sends whatever URI a real BMC would be given.
fn resolve_media(uri: &str) -> std::result::Result<String, String> {
    let path = if let Some(rest) = uri.strip_prefix("file://") {
        // `file:///abs/path` — the authority is empty for a local file.
        let rest = rest.strip_prefix("localhost").unwrap_or(rest);
        if !rest.starts_with('/') {
            return Err(format!(
                "This service resolves `file://` URIs with an absolute path; `{uri}` has none."
            ));
        }
        percent_decode(rest)
    } else if uri.starts_with('/') {
        uri.to_string()
    } else {
        let scheme = uri.split_once("://").map(|(s, _)| s).unwrap_or("<none>");
        return Err(format!(
            "This BMC fronts local KVM and mounts only `file://` URIs or absolute paths; \
             it does not fetch `{scheme}` media. A real BMC would."
        ));
    };
    if std::path::Path::new(&path).is_file() {
        Ok(path)
    } else {
        Err(format!(
            "No file exists at `{path}` on the host running this service."
        ))
    }
}

// ===========================================================================
// Unit tests — the PURE halves. The wire conformance lives in
// tests/redfish_server_conformance.rs, anchored to DMTF's own fixtures.
// ===========================================================================

#[cfg(test)]
mod tests {
    use super::*;

    /// The registry templates render with their arguments substituted — the check
    /// that a `MessageArgs` array and the `Message` string actually agree.
    #[test]
    fn registry_messages_render_their_arguments() {
        assert_eq!(
            ACTION_PARAMETER_VALUE_NOT_IN_LIST.render(&[
                "Reboot",
                prop::RESET_TYPE,
                wire::ACTION_RESET
            ]),
            "The value 'Reboot' for the parameter ResetType in the action \
             #ComputerSystem.Reset is not in the list of acceptable values."
        );
        assert_eq!(
            RESOURCE_MISSING_AT_URI.render(&["file:///nope.iso"]),
            "The resource at the URI 'file:///nope.iso' was not found."
        );
        // A zero-arg message renders verbatim.
        assert_eq!(NO_VALID_SESSION.render(&[]), NO_VALID_SESSION.template);
        // No `%N` may survive a full substitution.
        for m in ALL_MESSAGES {
            let args: Vec<&str> = (0..m.nargs).map(|_| "X").collect();
            let out = m.render(&args);
            assert!(!out.contains('%'), "{} left a placeholder: {out}", m.id);
        }
    }

    #[test]
    fn message_ids_are_registry_qualified() {
        assert_eq!(
            RESOURCE_MISSING_AT_URI.message_id(),
            "Base.1.19.0.ResourceMissingAtURI"
        );
        for m in ALL_MESSAGES {
            assert!(
                m.message_id().starts_with("Base.1.19.0."),
                "{} is unqualified",
                m.id
            );
        }
    }

    /// The routing table is built from the same `wire` functions the client builds
    /// its URLs from. RED-when-broken: hand-type any route and this fails.
    #[test]
    fn every_route_is_the_path_the_client_would_ask_for() {
        let cfg = NodeConfig::new("System.Embedded.1").media_slot("CD");
        let r = Routes::for_node(&cfg);
        assert_eq!(r.system, "/redfish/v1/Systems/System.Embedded.1");
        assert_eq!(
            r.reset,
            "/redfish/v1/Systems/System.Embedded.1/Actions/ComputerSystem.Reset"
        );
        assert_eq!(
            r.vm_insert,
            "/redfish/v1/Systems/System.Embedded.1/VirtualMedia/CD/Actions/VirtualMedia.InsertMedia"
        );
        assert_eq!(
            r.vm_eject,
            "/redfish/v1/Systems/System.Embedded.1/VirtualMedia/CD/Actions/VirtualMedia.EjectMedia"
        );
        assert_eq!(r.vm, "/redfish/v1/Systems/System.Embedded.1/VirtualMedia/CD");
        assert_eq!(
            r.vm_collection,
            "/redfish/v1/Systems/System.Embedded.1/VirtualMedia"
        );
    }

    #[test]
    fn media_resolution_refuses_by_name() {
        // A remote scheme is named, not silently swallowed.
        let e = resolve_media("https://depot/x.iso").unwrap_err();
        assert!(e.contains("https"), "names the scheme: {e}");
        // A path that is not there is named.
        let e = resolve_media("/nonexistent/never-built.iso").unwrap_err();
        assert!(e.contains("/nonexistent/never-built.iso"), "{e}");
        // A directory is not a medium.
        assert!(resolve_media("/tmp").is_err());
        // A real file resolves, by path and by file:// URI.
        let real = std::env::current_exe().unwrap();
        let real = real.to_string_lossy().into_owned();
        assert_eq!(resolve_media(&real).unwrap(), real);
        assert_eq!(resolve_media(&format!("file://{real}")).unwrap(), real);
        // Percent-encoding is DECODED, so a media path with a space or a non-ASCII
        // character (this box keeps its ISOs under `Hämtningar`) resolves.
        let odd = std::env::temp_dir().join(format!("draupnir rf-server å-{}.iso", std::process::id()));
        std::fs::write(&odd, b"medium").unwrap();
        let encoded = odd
            .to_string_lossy()
            .replace(' ', "%20")
            .replace('å', "%C3%A5");
        assert_eq!(
            resolve_media(&format!("file://{encoded}")).unwrap(),
            odd.to_string_lossy()
        );
        let _ = std::fs::remove_file(&odd);
    }

    #[test]
    fn percent_decode_handles_utf8_and_leaves_junk_alone() {
        assert_eq!(percent_decode("/H%C3%A4mtningar"), "/Hämtningar");
        assert_eq!(percent_decode("/a%zz"), "/a%zz", "invalid escape survives");
        assert_eq!(percent_decode("/plain"), "/plain");
    }

    #[test]
    fn basename_is_the_last_segment() {
        assert_eq!(basename("file:///a/b/c.iso").as_deref(), Some("c.iso"));
        assert_eq!(basename("x.iso").as_deref(), Some("x.iso"));
        assert_eq!(basename("/a/b/").as_deref(), None);
    }

    /// The node advertises exactly the targets it can apply, plus `None`. A token in
    /// `BOOT_TARGET_ALLOWABLE` that maps to no `BootOrder` must not appear here.
    #[test]
    fn the_node_honours_only_the_targets_it_can_actually_apply() {
        let honoured = boot_targets_this_node_honours();
        assert!(honoured.contains(&"Cd"));
        assert!(honoured.contains(&"Hdd"));
        assert!(honoured.contains(&"None"));
        assert!(!honoured.contains(&"Pxe"), "no PXE on a KVM front end");
        assert!(!honoured.contains(&"BiosSetup"));
    }

    /// A backend that records the spec it was handed and never launches anything —
    /// the same shape the conformance suite uses.
    #[derive(Default)]
    struct Recorder {
        specs: Mutex<Vec<BootSpec>>,
    }
    impl Boot for Recorder {
        fn boot(&self, spec: &BootSpec) -> Result<Machine> {
            self.specs.lock().unwrap().push(spec.clone());
            Ok(Machine::started(format!("rec-{}", spec.name), spec))
        }
    }
    impl Lifecycle for Recorder {
        fn power_on(&self, _m: &Machine) -> Result<()> {
            Ok(())
        }
        fn power_off(&self, _m: &Machine) -> Result<()> {
            Ok(())
        }
        fn status(&self, _m: &Machine) -> Result<PowerState> {
            Ok(PowerState::On)
        }
    }

    fn service_with(cfg: NodeConfig) -> Service {
        Service {
            routes: Routes::for_node(&cfg),
            cfg,
            state: Arc::new(Mutex::new(NodeState::fresh())),
            backend: Arc::new(Recorder::default()),
        }
    }

    /// **The translation, proven at the spec level.** A `Cd` override with media in
    /// the tray produces a spec that boots OFF the medium; an `Hdd` override produces
    /// one that carries NO medium at all.
    #[test]
    fn the_boot_override_selects_which_spec_the_node_boots() {
        let iso = std::env::current_exe().unwrap().to_string_lossy().into_owned();
        let svc = service_with(
            NodeConfig::new("s1")
                .local_disk("/var/lib/node.qcow2")
                .sized(2048, 4),
        );
        {
            let mut s = svc.state.lock().unwrap();
            s.image_uri = Some(iso.clone());
            s.image_path = Some(iso.clone());
            s.inserted = true;
            s.override_enabled = wire::OVERRIDE_ONCE.into();
            s.override_target = Some(BootTarget::Cd);
        }
        let spec = svc.spec_for(&svc.state.lock().unwrap()).unwrap();
        assert_eq!(spec.boot_order, BootOrder::Medium);
        assert_eq!(spec.medium_path(), Some(iso.as_str()));
        assert_eq!(spec.mem_mb, 2048);
        assert_eq!(spec.cores, 4);

        // Flip ONLY the override. The media stays in the tray, exactly as it would on
        // a real machine — and the spec must still stop naming it.
        svc.state.lock().unwrap().override_target = Some(BootTarget::Hdd);
        let spec = svc.spec_for(&svc.state.lock().unwrap()).unwrap();
        assert_eq!(spec.boot_order, BootOrder::Disk);
        assert_eq!(
            spec.medium_path(),
            None,
            "an Hdd override must not leave the installer medium attached — that is \
             how a boot override gets accepted without ever being applied"
        );
    }

    #[test]
    fn an_override_the_node_cannot_apply_is_refused_by_name_not_coerced() {
        let svc = service_with(NodeConfig::new("s1"));
        let req = Request {
            method: "PATCH".into(),
            path: wire::system_path("s1"),
            headers: BTreeMap::new(),
            body: serde_json::to_vec(&json!({
                "Boot": { "BootSourceOverrideTarget": "Pxe" }
            }))
            .unwrap(),
        };
        let resp = svc.patch_system(&req);
        assert_eq!(resp.status, 400);
        let v: Value = serde_json::from_slice(&resp.body).unwrap();
        assert_eq!(v["error"]["code"], "Base.1.19.0.PropertyValueNotInList");
        assert!(
            v["error"]["message"].as_str().unwrap().contains("Pxe"),
            "the refusal quotes the value: {v}"
        );
        // ...and nothing was applied.
        assert_eq!(svc.state.lock().unwrap().override_target, None);
    }

    #[test]
    fn a_cd_override_with_an_empty_tray_refuses_instead_of_booting_something_else() {
        let svc = service_with(NodeConfig::new("s1").local_disk("/var/lib/node.qcow2"));
        {
            let mut s = svc.state.lock().unwrap();
            s.override_enabled = wire::OVERRIDE_ONCE.into();
            s.override_target = Some(BootTarget::Cd);
        }
        let err = svc.spec_for(&svc.state.lock().unwrap()).unwrap_err();
        assert!(err.contains("no virtual media is inserted"), "{err}");
        // The configured disk is NOT quietly substituted.
        assert!(!err.contains("qcow2"), "{err}");
    }

    #[test]
    fn a_hdd_override_on_a_diskless_node_refuses_instead_of_falling_back_to_the_medium() {
        let iso = std::env::current_exe().unwrap().to_string_lossy().into_owned();
        let svc = service_with(NodeConfig::new("s1")); // no local disk
        {
            let mut s = svc.state.lock().unwrap();
            s.image_path = Some(iso);
            s.inserted = true;
            s.override_enabled = wire::OVERRIDE_ONCE.into();
            s.override_target = Some(BootTarget::Hdd);
        }
        let err = svc.spec_for(&svc.state.lock().unwrap()).unwrap_err();
        assert!(err.contains("no local disk"), "{err}");
    }

    /// A one-time override is CONSUMED by the boot that applied it.
    #[test]
    fn a_once_override_does_not_steer_the_next_boot_too() {
        let iso = std::env::current_exe().unwrap().to_string_lossy().into_owned();
        let svc = service_with(NodeConfig::new("s1"));
        {
            let mut s = svc.state.lock().unwrap();
            s.image_uri = Some(iso.clone());
            s.image_path = Some(iso);
            s.inserted = true;
            s.override_enabled = wire::OVERRIDE_ONCE.into();
            s.override_target = Some(BootTarget::Cd);
        }
        assert_eq!(svc.power_up(wire::RESET_ON).status, 204);
        let s = svc.state.lock().unwrap();
        assert_eq!(s.override_enabled, wire::OVERRIDE_DISABLED);
        assert_eq!(s.override_target, None);
        assert_eq!(s.last_spec.as_ref().unwrap().boot_order, BootOrder::Medium);
    }

    #[test]
    fn a_bad_reset_type_is_refused_and_nothing_boots() {
        let svc = service_with(NodeConfig::new("s1"));
        let req = Request {
            method: "POST".into(),
            path: wire::reset_path("s1"),
            headers: BTreeMap::new(),
            body: serde_json::to_vec(&json!({ "ResetType": "Reboot" })).unwrap(),
        };
        let resp = svc.post_reset(&req);
        assert_eq!(resp.status, 400);
        let v: Value = serde_json::from_slice(&resp.body).unwrap();
        assert_eq!(
            v["error"]["code"],
            "Base.1.19.0.ActionParameterValueNotInList"
        );
        assert_eq!(v["error"]["@Message.ExtendedInfo"][0]["MessageArgs"][0], "Reboot");
        assert!(svc.state.lock().unwrap().machine.is_none(), "nothing booted");
    }

    #[test]
    fn a_reset_with_no_reset_type_names_the_missing_parameter() {
        let svc = service_with(NodeConfig::new("s1"));
        let req = Request {
            method: "POST".into(),
            path: wire::reset_path("s1"),
            headers: BTreeMap::new(),
            body: b"{}".to_vec(),
        };
        let resp = svc.post_reset(&req);
        assert_eq!(resp.status, 400);
        let v: Value = serde_json::from_slice(&resp.body).unwrap();
        assert_eq!(v["error"]["code"], "Base.1.19.0.ActionParameterMissing");
        assert!(v["error"]["message"].as_str().unwrap().contains("ResetType"));
    }

    #[test]
    fn insert_media_pointing_at_nothing_is_refused_by_name() {
        let svc = service_with(NodeConfig::new("s1"));
        let req = Request {
            method: "POST".into(),
            path: svc.routes.vm_insert.clone(),
            headers: BTreeMap::new(),
            body: serde_json::to_vec(&wire::insert_media_body("/nonexistent/never-built.iso"))
                .unwrap(),
        };
        let resp = svc.post_insert_media(&req);
        assert_eq!(resp.status, 400);
        let v: Value = serde_json::from_slice(&resp.body).unwrap();
        assert_eq!(v["error"]["code"], "Base.1.19.0.ResourceMissingAtURI");
        assert!(v["error"]["message"]
            .as_str()
            .unwrap()
            .contains("/nonexistent/never-built.iso"));
        // Nothing was mounted.
        assert!(!svc.state.lock().unwrap().inserted);
    }

    #[test]
    fn a_method_the_resource_does_not_take_answers_405_with_allow() {
        let svc = service_with(NodeConfig::new("s1").credentials("admin", "pw"));
        let auth = wire::basic_auth_header("admin", "pw");
        let mut headers = BTreeMap::new();
        headers.insert("authorization".to_string(), auth);
        // GET on an action target.
        let req = Request {
            method: "GET".into(),
            path: wire::reset_path("s1"),
            headers: headers.clone(),
            body: Vec::new(),
        };
        let resp = svc.dispatch(&req);
        assert_eq!(resp.status, 405);
        assert!(resp
            .headers
            .iter()
            .any(|(k, v)| k == "Allow" && v == "POST"));
        // POST on the ComputerSystem.
        let req = Request {
            method: "POST".into(),
            path: wire::system_path("s1"),
            headers,
            body: Vec::new(),
        };
        let resp = svc.dispatch(&req);
        assert_eq!(resp.status, 405);
        assert!(resp
            .headers
            .iter()
            .any(|(k, v)| k == "Allow" && v == "GET, PATCH"));
    }

    #[test]
    fn everything_below_the_service_root_needs_a_credential() {
        let svc = service_with(NodeConfig::new("s1").credentials("admin", "pw"));
        let unauth = |path: &str| Request {
            method: "GET".into(),
            path: path.into(),
            headers: BTreeMap::new(),
            body: Vec::new(),
        };
        // The two DSP0266 unauthenticated URIs are reachable...
        assert_eq!(svc.dispatch(&unauth(wire::PROTOCOL_VERSION_PATH)).status, 200);
        assert_eq!(svc.dispatch(&unauth(wire::SERVICE_ROOT_PATH)).status, 200);
        // ...and nothing else is.
        for p in [
            wire::SYSTEMS_PATH,
            &wire::system_path("s1"),
            &wire::virtual_media_path("s1", "CD"),
        ] {
            let r = svc.dispatch(&unauth(p));
            assert_eq!(r.status, 401, "{p}");
            assert!(r.headers.iter().any(|(k, _)| k == "WWW-Authenticate"), "{p}");
        }
        // A WRONG password is not a credential either.
        let mut headers = BTreeMap::new();
        headers.insert(
            "authorization".into(),
            wire::basic_auth_header("admin", "wrong"),
        );
        let req = Request {
            method: "GET".into(),
            path: wire::system_path("s1"),
            headers,
            body: Vec::new(),
        };
        assert_eq!(svc.dispatch(&req).status, 401);
    }

    #[test]
    fn an_unknown_uri_is_a_dmtf_shaped_404_naming_the_uri() {
        let svc = service_with(NodeConfig::new("s1").credentials("admin", ""));
        let mut headers = BTreeMap::new();
        headers.insert("authorization".into(), wire::basic_auth_header("admin", ""));
        let req = Request {
            method: "GET".into(),
            path: "/redfish/v1/Systems/other".into(),
            headers,
            body: Vec::new(),
        };
        let resp = svc.dispatch(&req);
        assert_eq!(resp.status, 404);
        let v: Value = serde_json::from_slice(&resp.body).unwrap();
        assert_eq!(v["error"]["code"], "Base.1.19.0.ResourceNotFound");
        assert!(v["error"]["message"]
            .as_str()
            .unwrap()
            .contains("/redfish/v1/Systems/other"));
    }

    #[test]
    fn malformed_json_is_reported_as_malformed_json() {
        let svc = service_with(NodeConfig::new("s1"));
        let req = Request {
            method: "POST".into(),
            path: wire::reset_path("s1"),
            headers: BTreeMap::new(),
            body: b"{not json".to_vec(),
        };
        let resp = svc.post_reset(&req);
        assert_eq!(resp.status, 400);
        let v: Value = serde_json::from_slice(&resp.body).unwrap();
        assert_eq!(v["error"]["code"], "Base.1.19.0.MalformedJSON");
    }

    #[test]
    fn a_freshly_started_node_reports_itself_off_with_an_empty_tray() {
        let svc = service_with(NodeConfig::new("s1"));
        let sys = svc.computer_system();
        assert_eq!(sys["PowerState"], "Off");
        assert_eq!(sys["Boot"]["BootSourceOverrideEnabled"], "Disabled");
        assert_eq!(sys["Boot"]["BootSourceOverrideTarget"], "None");
        let vm = svc.virtual_media();
        assert_eq!(vm["Inserted"], false);
        assert!(vm["Image"].is_null());
    }
}