bugwarden 0.6.0

//! bugwarden — Rust MCP server for Bugzilla with operator-controlled
//! security guards.
//!
//! The binary is a thin transport wrapper; the CLI and the MCP tool surface
//! live in the `bugwarden` library crate (`config`, `server`) so integration
//! tests can drive the tools without a process boundary.

use std::sync::atomic::Ordering;
use std::sync::{Arc, OnceLock};

use anyhow::Context;
use bugwarden::audit::{
    policy_hash_of, select_sinks, AuditConfig, AuditSink, AuditState, FailMode, SinkSelection,
    TransportKind,
};
use bugwarden::http_auth::{self, HttpEnv};
use bugwarden::http_session::AuditedSessionManager;
use bugwarden::otel::{self, OtelEnv, Pipeline};
use bugwarden::stdio::DiscoverAnswering;
use bugwarden::tracing_fields::CappedFields;
use bugwarden::{config, server};
use bugwarden_core::{guard::Guard, policy::Policy};
use clap::Parser;
use rmcp::{transport::streamable_http_server::StreamableHttpService, ServiceExt};
use tracing_subscriber::layer::SubscriberExt as _;
use tracing_subscriber::util::SubscriberInitExt as _;
use tracing_subscriber::EnvFilter;

use config::{Cli, Transport};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let cli = Cli::parse();

    // OTLP export configuration, resolved before anything is installed:
    // parsing the environment starts no task and opens no socket, but a
    // protocol this build cannot speak has to be a startup error rather
    // than a surprise once records exist. Unset or empty
    // OTEL_EXPORTER_OTLP_ENDPOINT means the whole feature stays off.
    let otel_config = otel::resolve(&OtelEnv::from_env())?;
    // Filled in below, once the audit sink has opened; the diagnostics
    // layer reads it and does nothing while it is empty.
    let otel_slot: Arc<OnceLock<Arc<Pipeline>>> = Arc::new(OnceLock::new());

    // Tracing always goes to stderr: stdout belongs to the stdio transport.
    // The OTLP layer, when export is on, sits beside the stderr one under
    // the same filter, so both carry the same events.
    //
    // `fmt_fields` is the bound on what a client can put there (#260,
    // #266, #275): every field of every line, ours and rmcp's alike, is
    // cut at 1024 chars and has every control character and every
    // mandatory line break escaped — C0, DEL, C1 and U+2028/U+2029
    // — whatever this filter says. LF is in that set, so a field can
    // no longer end this line and open one of its own.
    let registry = tracing_subscriber::registry()
        .with(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")))
        .with(
            tracing_subscriber::fmt::layer()
                .fmt_fields(CappedFields)
                .with_writer(std::io::stderr)
                .with_ansi(false),
        );
    if otel_config.is_some() {
        registry
            .with(Pipeline::diagnostics_layer(otel_slot.clone()))
            .init();
    } else {
        registry.init();
    }
    // On both branches: the hook routes a panic into tracing instead of
    // letting std write the payload to fd 2, which over stdio is the
    // peer's pipe (#270). After the subscriber rather than before it not
    // because the order changes where a later panic goes — the dispatcher
    // is read at panic time — but so that a panic in the window above,
    // during CLI parsing or OTLP resolution, is still printed by std's
    // hook instead of vanishing into a subscriber that does not exist.
    bugwarden::panic_hook::install();

    // The http bearer gate, resolved before anything else runs: over http
    // the port is the access boundary, and a half-configured credential must
    // never degrade into open access. Ahead of the policy load, the Bugzilla
    // client, the identity preflight and the audit sink alike, so a token
    // misconfiguration aborts startup without binding a port, opening a
    // socket to Bugzilla, or creating or rotating an audit file. `None` for
    // stdio, where the client owns the process and the tokens are ignored.
    let http_auth =
        http_auth::resolve_for(cli.transport, &HttpEnv::from_env(), cli.insecure_no_auth)
            .map(|auth| auth.map(Arc::new))?;
    if let Some(auth) = &http_auth {
        auth.log_startup_mode();
    }

    // Which audit sinks this deployment runs (issue #31, revised
    // 2026-08-18): file, OTLP, both, or none. Resolved here — pure
    // configuration, no file touched — so the two ambiguous spellings
    // (an endpoint with no file decision, `none` with no endpoint) refuse
    // startup before any network or filesystem work. The sinks themselves
    // open after the preflight, below, exactly as before.
    let sinks = select_sinks(cli.audit_config.as_deref(), otel_config.is_some())?;

    // Guard policy comes ONLY from the TOML file given at startup (I1);
    // without one the built-in default policy applies.
    let mut policy = match &cli.policy {
        Some(path) => Policy::load(path)
            .with_context(|| format!("failed to load guard policy from {}", path.display()))?,
        None => Policy::default(),
    };
    // CLI/env can only tighten policy (I9).
    policy.global.read_only |= cli.read_only;

    // API key custody (stdio-without-key bail, http --api-key warn, key
    // file handling) is resolved inside BugWarden::new — before the audit
    // sink below, so a key misconfiguration aborts startup without first
    // creating or rotating an audit file.
    // Built in the library target, not here: the client's identity to
    // Bugzilla (I12-adjacent, issue #55) is only testable where a test can
    // reach the constructor.
    let bz = Arc::new(server::bugzilla_client(&cli)?);
    let guard = Arc::new(Guard { policy });
    let cfg = Arc::new(cli);
    let server =
        server::BugWarden::new(cfg.clone(), guard, bz).context("failed to build the MCP server")?;
    // Turns a silent whoami-blackout (I4 denying every created_by_me
    // classification) into an actionable startup failure. Before the audit
    // sink below: a failed preflight must not create or rotate an audit
    // file (the same ordering rationale as key custody resolution above).
    server.preflight().await?;

    // OTLP export, started AFTER the identity preflight and BEFORE the
    // audit file is created: a collector that will not take records must
    // refuse to start without leaving a file behind, the same ordering
    // the preflight already keeps. Off entirely when no endpoint is
    // configured: no task, no thread, no layer.
    let otel = match otel_config {
        Some(otel_cfg) => Some(Arc::new(Pipeline::start(otel_cfg)?)),
        None => None,
    };
    if let Some(pipeline) = &otel {
        // Fills the slot the diagnostics layer installed above reads.
        let _ = otel_slot.set(pipeline.clone());
        // The delivery probe, mirroring the identity preflight: a
        // deployment that cannot deliver its audit records refuses to
        // start (bounded retry inside — a co-starting collector may lose
        // the race by a few seconds).
        pipeline.probe().await?;
        tracing::info!(
            "OTLP export enabled: audit records are exported to the configured \
             collector and gate serving when delivery fails; diagnostics are \
             copied best-effort"
        );
    }

    // Audit sinks, per the selection above. The fail mode falls back to
    // the transport-derived default — an OTLP-only sink has no document
    // to say otherwise, so it always derives — and the policy digest ties
    // every record to the policy document in force.
    let audit_sink = match sinks {
        SinkSelection::NoAudit => None,
        SinkSelection::FileOnly | SinkSelection::Both | SinkSelection::OtlpOnly => {
            let audit_cfg = match sinks {
                SinkSelection::FileOnly | SinkSelection::Both => {
                    match cfg.audit_config.as_deref() {
                        Some(path) => AuditConfig::load(path)?,
                        // Unreachable: `select_sinks` returns a file-bearing
                        // selection only for a real path. Refuse rather than
                        // panic, like the http bearer gate below.
                        None => anyhow::bail!(
                            "internal error: a file-bearing sink selection has no path"
                        ),
                    }
                }
                _ => AuditConfig::fileless(),
            };
            let transport = match cfg.transport {
                Transport::Stdio => TransportKind::Stdio,
                Transport::Http => TransportKind::Http,
            };
            let fail_mode = FailMode::resolve(audit_cfg.fail_mode, transport);
            // The digest re-reads the file Policy::load parsed above
            // (load also owns the unix permission warning, so it stays
            // the one parse path). A rewrite between the two reads could
            // hash different bytes than were parsed; the window is
            // operator-local and harmless — the operator who swaps the
            // policy mid-startup gets the digest of what is on disk.
            let policy_hash = match &cfg.policy {
                Some(policy_path) => {
                    let bytes = std::fs::read(policy_path).with_context(|| {
                        format!(
                            "failed to read guard policy from {} for the audit digest",
                            policy_path.display()
                        )
                    })?;
                    Some(policy_hash_of(&bytes))
                }
                None => None,
            };
            let sink = AuditSink::open(audit_cfg).context("failed to open the audit sink")?;
            Some((sink, fail_mode, policy_hash))
        }
    };

    let audit = audit_sink.map(|(sink, fail_mode, policy_hash)| {
        let sink = match &otel {
            Some(pipeline) => sink.with_export(pipeline.audit_exporter()),
            None => sink,
        };
        Arc::new(AuditState::new(sink, fail_mode, policy_hash))
    });
    if audit.is_none() && cfg.transport == Transport::Http {
        tracing::warn!(
            "auditing is OFF: remote tool calls over http will leave no audit \
             record — pass --audit-config / BUGWARDEN_AUDIT_CONFIG for a file, \
             or set OTEL_EXPORTER_OTLP_ENDPOINT with BUGWARDEN_AUDIT_CONFIG=none \
             for a collector-only trail"
        );
    }

    let server = match audit {
        Some(audit) => server.with_audit(audit),
        None => server,
    };
    // Per-request scopes exist only where a per-request credential does:
    // authenticated http. stdio issues none, and --insecure-no-auth issues
    // none by definition.
    let server =
        server.with_scope_enforcement(http_auth.as_ref().is_some_and(|auth| !auth.is_insecure()));

    // Cloned into the serve future so a stdio SIGTERM can flush before
    // `process::exit`. The original stays here for the HTTP / peer-close
    // path, which returns from the future instead of exiting.
    let otel_on_signal = otel.clone();
    // The result is held rather than propagated so the OTLP flush below
    // runs on every exit path, a transport error included.
    let served: anyhow::Result<()> = async move {
        match cfg.transport {
            Transport::Stdio => {
                // Two stages: an unused stdio container sits in `serve`
                // (handshake). After initialize it sits in `waiting`.
                // Load-bearing order: `shutdown_signal` arms the handlers
                // in the call, so the startup line below cannot be read by
                // anything that then races an unhandled signal (#173).
                let shutdown = shutdown_signal();
                tokio::pin!(shutdown);
                // Bounded stdin, never rmcp's bare `stdio()`: it reads a
                // frame into an unbounded `Vec` (#234). Built here for the
                // same reason `http_server_config()` is built in the other
                // arm — the cap comes from `server`'s own policy, and
                // `serve` takes it by value.
                let (stdin, stdout) = server.stdio_transport();
                // rmcp maps the over-cap read error to `receive() -> None`,
                // which the service reports as an ordinary close, so this
                // flag is the only thing that tells the two apart below.
                let over_cap = stdin.over_cap();
                // `server/discover` is answered by the transport, never by
                // rmcp's lifecycle picker: rmcp commits the session to the
                // handshake-free lifecycle on the first non-`initialize`
                // frame, sticking a flag it offers no way to clear (#267).
                // Built before `serve` consumes `server`.
                let transport = DiscoverAnswering::framed(stdin, stdout, server.clone());
                // An answered probe leaves rmcp waiting for the first
                // COMMITTING frame, so a client that only probed and hung
                // up closes the stream inside `initialize`'s wait, where
                // rmcp reports `ConnectionClosed` rather than a clean end.
                let probed = transport.answered();
                tracing::info!("Starting Bugzilla MCP server on stdio");
                let service = tokio::select! {
                    result = server.serve(transport) => match result {
                        Ok(service) => service,
                        // A probe-only client is a peer hangup, not a
                        // serving failure: exit 0, as before #267. Not a
                        // refused frame though — that closes the same way
                        // and must stay a failure exit.
                        Err(rmcp::service::ServerInitializeError::ConnectionClosed(_))
                            if probed.load(Ordering::Acquire)
                                && !over_cap.load(Ordering::Acquire) =>
                        {
                            tracing::info!("peer hung up after a server/discover probe");
                            return Ok(());
                        }
                        // The cap fired, so THIS is the refusal and rmcp's
                        // error is its echo: bugwarden's bounded reader
                        // returned an error, rmcp turned that into
                        // `receive() -> None`, and `None` before the
                        // handshake is `ConnectionClosed` — a shape that
                        // says the peer left when the server refused. Same
                        // sentence as the post-handshake arm below, from
                        // the same constant, so one refusal keeps one
                        // description on either side of `initialize`.
                        Err(rmcp::service::ServerInitializeError::ConnectionClosed(_))
                            if over_cap.load(Ordering::Acquire) =>
                        {
                            tracing::error!("serving error: {OVER_CAP_CLOSE}");
                            anyhow::bail!("{OVER_CAP_CLOSE}");
                        }
                        // Never `e` itself, in either line: rmcp's error
                        // carries the client's whole first frame (#261).
                        Err(e) => {
                            let what = serve_failure(&e);
                            tracing::error!("serving error: {what}");
                            return Err(anyhow::anyhow!("stdio serving failed: {what}"));
                        }
                    },
                    () = &mut shutdown => {
                        tracing::info!("received shutdown signal");
                        // serve() is already blocked in tokio::io::stdin()'s
                        // uncancellable read. Returning from main drops the
                        // runtime onto that blocking thread. Flush first:
                        // process::exit would otherwise skip the tail of a
                        // load-bearing OTLP-only sink.
                        flush_otel_and_exit(otel_on_signal.as_deref()).await;
                    }
                };
                let cancel = service.cancellation_token();
                tokio::select! {
                    result = service.waiting() => {
                        result?;
                        // A refused frame closes the transport exactly like
                        // a peer hangup, and `waiting()` reports both as
                        // `Ok`. Without this the post-handshake half exits
                        // 0 while the pre-handshake half exits 1 — the same
                        // refusal, two different exit codes.
                        if over_cap.load(Ordering::Acquire) {
                            anyhow::bail!("{OVER_CAP_CLOSE}");
                        }
                    }
                    () = shutdown => {
                        tracing::info!("received shutdown signal");
                        cancel.cancel();
                        flush_otel_and_exit(otel_on_signal.as_deref()).await;
                    }
                }
            }
            Transport::Http => {
                let ct = tokio_util::sync::CancellationToken::new();

                // Derived from the server's own guard policy (the POST body cap
                // follows `global.max_attachment_bytes`, issue #52), so it is
                // built while `server` can still be borrowed.
                let config = server
                    .http_server_config()?
                    .with_cancellation_token(ct.child_token());
                // Load-bearing pair with `serve_http` in
                // tests/http_transport_wiremock.rs: the audit session id
                // comes from this wrapper's stamp and nowhere else, so a
                // plain LocalSessionManager here serves every http record
                // with no session id at all. The join test forces the
                // harness; only review forces this call site.
                let service = StreamableHttpService::new(
                    move || Ok(server.clone()),
                    AuditedSessionManager::default().into(),
                    config,
                );
                // `resolve_for` returns the gate for every http start, so the
                // bail below is unreachable — it exists so a future refactor
                // that lost the gate fails to serve rather than serving open.
                let Some(auth) = http_auth else {
                    anyhow::bail!("internal error: the http transport has no resolved bearer gate");
                };
                let router = http_auth::guard_router(
                    axum::Router::new().nest_service("/mcp", service),
                    auth,
                );
                let addr = format!("{}:{}", cfg.host, cfg.port);
                // Armed before the bind, so nothing that can reach the port
                // can outrun the handlers; see `shutdown_signal` (#173).
                let shutdown = shutdown_signal();
                let tcp_listener = tokio::net::TcpListener::bind(&addr)
                    .await
                    .with_context(|| format!("failed to bind {addr}"))?;
                // The bound address, not the requested one: `--port 0` asks
                // the kernel for a free port, and the operator (or a test)
                // has no other way to learn which.
                let bound = tcp_listener
                    .local_addr()
                    .with_context(|| format!("failed to read the local address of {addr}"))?;
                tracing::info!("Starting Bugzilla MCP server on {bound}");
                // Connect-info makes the remote peer address available in the
                // request extensions, where the audit session info reads it.
                axum::serve(
                    tcp_listener,
                    router.into_make_service_with_connect_info::<std::net::SocketAddr>(),
                )
                .with_graceful_shutdown(async move {
                    shutdown.await;
                    tracing::info!("received shutdown signal");
                    // Tear the live streamable-HTTP transport down with the
                    // listener: axum's graceful shutdown alone waits for
                    // in-flight connections, and an open MCP session is one.
                    ct.cancel();
                })
                .await?;
            }
        }
        Ok(())
    }
    .await;

    // Best-effort, bounded flush of whatever is still queued. On http this
    // runs after graceful shutdown, i.e. after the SIGINT that cancelled
    // it; on stdio after the peer closed the session.
    if let Some(pipeline) = &otel {
        pipeline.shutdown().await;
    }

    served
}

/// What an over-cap inbound frame did, in one sentence both refusal
/// sites use.
///
/// The bounded reader closes the transport the moment a frame passes the
/// cap, and rmcp reports that to the service the way it reports a peer
/// hangup: `receive() -> None`, which before the handshake becomes
/// `ServerInitializeError::ConnectionClosed` and after it an ordinary
/// `Ok` close. Both arms therefore have to say so themselves, and saying
/// it from one constant is what stops the same refusal acquiring two
/// descriptions on either side of `initialize`.
const OVER_CAP_CLOSE: &str = "stdio transport closed: an inbound frame exceeded the request cap";

/// A fixed, server-authored classification of a failed stdio handshake —
/// the only thing either exit path may say about one (#261).
///
/// rmcp 3.1.4's `ServerInitializeError`
/// (`service/server.rs:82-104`, `#[non_exhaustive]`) puts the client's
/// whole first frame into `ExpectedInitializeRequest`, and its
/// `#[error("expect initialized request, but received: {0:?}")]` puts it
/// into DISPLAY as well as `Debug` — so neither the error nor its Display
/// may be logged, and it may not be propagated into `main`'s
/// `anyhow::Result` either, since Rust prints `Error: {:?}` of that and
/// anyhow's `Debug` opens with the same Display. A 100 000-char `query` in
/// a pre-`initialize` `tools/call` measured 100 403 and 100 365 chars on
/// those two lines, bounded only by the transport's frame cap.
///
/// Every arm returns a literal, so both lines are bounded by this file. The
/// frame's KIND is named because it is the diagnostic an operator needs and
/// it comes from the enum, not from the frame; the method name is NOT,
/// because `ClientRequest::CustomRequest` (`model.rs:948`) holds a
/// free-form `method: String` and rmcp's untagged deserialization routes
/// every unrecognised method there — naming it would put client bytes back
/// into the line this exists to bound. `ConnectionClosed`'s payload is the
/// `&'static str` its one server-side construction site passes
/// (`service/server.rs:442, 511`), but it is typed `String` on a
/// `#[non_exhaustive]` enum, so it is classified rather than echoed.
///
/// What that costs, deliberately: `TransportError` used to show the
/// underlying `io::Error` (`Broken pipe (os error 32)`), which is
/// server-side I/O and carries no client bytes. It is dropped anyway,
/// because a per-variant exception is the thing that erodes — the payload
/// is a `DynamicTransportError` whose contents are the transport's to
/// choose, on an enum upstream may extend.
///
/// The frame kinds are matched in their own nested `match` with no arm of
/// its own for the unexpected, so a `JsonRpcMessage` that grew a fifth
/// kind would fail to compile here rather than fall quietly into the
/// wildcard below — which exists only for `ServerInitializeError`'s
/// `#[non_exhaustive]` and would otherwise swallow both.
fn serve_failure(error: &rmcp::service::ServerInitializeError) -> &'static str {
    use rmcp::model::ClientJsonRpcMessage as Frame;
    use rmcp::service::ServerInitializeError as Failure;

    match error {
        Failure::ExpectedInitializeRequest(Some(frame)) => match frame {
            // A request here is never `initialize` (it would have been
            // served) and never `ping` (rmcp answers those in its
            // pre-init loop).
            Frame::Request(_) => "the first frame was a request other than initialize",
            Frame::Notification(_) => "the first frame was a notification, not initialize",
            Frame::Response(_) => "the first frame was a response, not initialize",
            Frame::Error(_) => "the first frame was an error reply, not initialize",
        },
        // Unreachable on rmcp 3.1.4: both construction sites pass `Some`.
        Failure::ExpectedInitializeRequest(None) => "no first frame was received",
        // Actor-neutral on purpose: an EOF, a read error and this build's
        // own frame-cap refusal all arrive here as the same `None`, and
        // the arm cannot tell them apart. The caller names the cap when it
        // knows; rmcp logs a read error itself.
        Failure::ConnectionClosed(_) => "the stream ended before initialize",
        // Unreachable too: rmcp builds the value it checks here by mapping
        // this handler's own `InitializeResult` into `ServerResult`.
        Failure::UnexpectedInitializeResponse(_) => {
            "this server answered initialize with the wrong result"
        }
        Failure::InitializeFailed(_) => "this server refused initialize",
        Failure::TransportError { .. } => "the stdio transport failed during the handshake",
        // Unreachable as long as `ServiceExt::serve` makes the cancellation
        // token itself and never cancels it.
        Failure::Cancelled => "the handshake was cancelled",
        _ => "the handshake failed",
    }
}

/// Bounded OTLP flush, then `_exit`. Used on the stdio signal arms: those
/// cannot return from `main` (rmcp's stdin read is uncancellable) and
/// must not skip a load-bearing collector either.
async fn flush_otel_and_exit(otel: Option<&Pipeline>) -> ! {
    if let Some(pipeline) = otel {
        pipeline.shutdown().await;
    }
    std::process::exit(0);
}

/// Arm the shutdown signals; the returned future resolves when one arrives.
///
/// `SIGINT` and, on Unix, `SIGTERM`. As container PID 1 the kernel does not
/// apply SIGTERM's default terminate action, so without this waiter `docker
/// stop` / `podman stop` wait out the runtime grace period and SIGKILL
/// (issue #114). Both signals take the same path: the caller then cancels
/// whatever is serving (the HTTP transport token, or the stdio `select!`).
///
/// Arming happens in the call, not on the first poll of the returned future
/// — an `async fn` body runs nothing until polled, so a signal delivered in
/// the gap between "serving" and that first poll still got the kernel's
/// default action, which is a kill rather than the graceful exit this exists
/// for (#173). Both callers therefore call this *before* they announce
/// readiness, which is what lets anything waiting on the startup line signal
/// immediately. Arm no earlier than that needs: everything ahead of the call
/// (policy load, preflight, OTLP probe) keeps the default disposition, so a
/// signal during a slow startup still stops the process at once.
///
/// A signal whose handler will not install is logged and dropped from the
/// set — refusing to start would be worse than a container that still needs
/// `--init`. If neither installs, the future stays pending: resolving would
/// shut down a healthy server instead of failing to stop a doomed one.
#[cfg(unix)]
fn shutdown_signal() -> impl std::future::Future<Output = ()> {
    use tokio::signal::unix::{signal, Signal, SignalKind};

    fn arm(kind: SignalKind, name: &str) -> Option<Signal> {
        signal(kind)
            .inspect_err(|err| {
                tracing::error!(
                    error = %err,
                    "failed to install the {name} handler; {name} will not stop the process"
                );
            })
            .ok()
    }

    let interrupt = arm(SignalKind::interrupt(), "SIGINT");
    let terminate = arm(SignalKind::terminate(), "SIGTERM");
    async move {
        match (interrupt, terminate) {
            (Some(mut interrupt), Some(mut terminate)) => {
                tokio::select! {
                    _ = interrupt.recv() => {}
                    _ = terminate.recv() => {}
                }
            }
            (Some(mut only), None) | (None, Some(mut only)) => {
                only.recv().await;
            }
            (None, None) => std::future::pending::<()>().await,
        }
    }
}

/// Non-Unix fallback: ctrl-c only, and installed on the first poll — the
/// arming contract above is Unix-only.
#[cfg(not(unix))]
fn shutdown_signal() -> impl std::future::Future<Output = ()> {
    async {
        let _ = tokio::signal::ctrl_c().await;
    }
}