gunnar-grpc 1.5.0

//! `Admin`. Health, stats, storage info, and the capacity question.
//!
//! `Health` reports the v1 auth hole **on the wire**: `grpc_authentication_
//! enabled` is `false` and `grpc_bind` says exactly where the unauthenticated
//! surface is listening. A hole documented only in prose is a hole nobody
//! checks; a field a UI can render is one somebody notices.

use gunnar_proto::admin_server::Admin;
use gunnar_proto::{
    CapacityRequest, CapacityResponse, GrowResumeRequest, GrowStartRequest, GrowStatusRequest,
    GrowStatusResponse, HealthRequest, HealthResponse, IndexProgress, RepoPath, StatsRequest,
    StatsResponse, StorageInfoRequest, StorageInfoResponse, VerifyChainRequest, VerifyChainResponse,
};
use tonic::{Request, Response, Status};

use crate::auth::Access;
use crate::service::AdminService;
use crate::status::opt_repo_path;
use gunnar_proto::GrantLevel;

#[tonic::async_trait]
impl Admin for AdminService {
    async fn health(
        &self,
        _request: Request<HealthRequest>,
    ) -> Result<Response<HealthResponse>, Status> {
        // Health is authenticated too. It reports the bind addresses of both
        // listeners and whether authentication is on, which is precisely the
        // reconnaissance an unauthenticated probe wants, and a liveness check
        // on a loopback control plane can hold a key like anything else.
        self.ctx
            .authorize(&_request, Access::Server, GrantLevel::Read, "Admin.Health")?;
        // Taken once, so every field below describes the same instant. Two
        // `report()` calls could straddle an incoming v1 credential and
        // report a count that no roster explains.
        let unbound = self.ctx.unbound.report();
        Ok(Response::new(HealthResponse {
            serving: true,
            agent: gunnar_proto::AGENT.to_string(),
            version: env!("CARGO_PKG_VERSION").to_string(),
            // **What is actually running**, beside the crate version that is
            // pinned in a manifest. `gunnar control status` has printed this as
            // `build` since it existed; Health never carried it, so every
            // consumer that had only this message — the console's header band
            // among them — could show `1.5.0` for a box 228 commits past the
            // tag and be believed. `unknown` is normalised to empty so a reader
            // gets "not stamped" rather than a word that looks like a ref.
            build_ref: match gunnar_common::BUILD_REF {
                "unknown" => String::new(),
                stamped => stamped.to_string(),
            },
            uptime_seconds: self.ctx.started.elapsed().as_secs(),
            // The RUNTIME answer, not the build-time constant: a build that
            // has authentication and a listener serving allow-all are
            // different facts, and an operator needs the second one.
            grpc_authentication_enabled: self.ctx.config.authentication_enabled(),
            grpc_bind: self.ctx.config.bind.to_string(),
            ssh_bind: self.ctx.config.ssh_bind_hint.clone(),
            grpc_mutual_tls: self.ctx.config.mutual_tls,
            // Every gate refuses the method-blind credential (5.3.3); only a
            // listener with no gate at all (allow-all) does not.
            grpc_require_bound_credentials: self.ctx.config.authentication_enabled(),
            // Read out of the SAME log the gate writes — `router_with_trust`
            // hands the gate this context's `Arc`. A second counter here
            // would be a second opinion about which clients still sign v1.
            unbound_credentials_accepted: unbound.accepted,
            unbound_credentials_refused: unbound.refused,
            unbound_credential_last_seen_unix_ms: unbound.last_seen_unix_ms,
            unbound_credential_callers_dropped: unbound.distinct_dropped,
            unbound_credential_callers: unbound
                .callers
                .into_iter()
                .map(|row| gunnar_proto::UnboundCredential {
                    fingerprint: row.fingerprint,
                    principal_id: row.principal_id,
                    method: row.method,
                    first_seen_unix_ms: row.first_seen_unix_ms,
                    last_seen_unix_ms: row.last_seen_unix_ms,
                    count: row.count,
                    refused: row.refused,
                })
                .collect(),
            server_unix_ms: estate_clock::now_unix_ms(),
            clock_boot: estate_clock::report::boot().map(clock_sync),
            clock_last: estate_clock::report::last().map(clock_sync),
            clock_skew_syncs_suppressed: estate_clock::trigger::suppressed(),
            // Read off the disk at each call: the journal is the boot stack's
            // and changes only across a boot, so there is nothing to cache.
            grow_state: crate::grow_journal::Summary::measure().map(|s| s.to_wire()),
            twins: self.twin_rows(),
            primary: self.ctx.control.primary_heartbeat(self.ctx.heartbeats.bounds().silent_after_ms),
            alarms: self.ctx.control.health_alarms(),
            status: if self.ctx.control.maintenance_reasons().is_empty() { gunnar_proto::HEALTH_SERVING.to_string() } else { gunnar_proto::HEALTH_MAINTENANCE.to_string() },
        }))
    }

    async fn stats(
        &self,
        _request: Request<StatsRequest>,
    ) -> Result<Response<StatsResponse>, Status> {
        self.ctx
            .authorize(&_request, Access::Server, GrantLevel::Read, "Admin.Stats")?;
        let counts = self.ctx.control.counts()?;
        let extras = self.ctx.control.stats_extras()?;
        let bus = self.ctx.bus.stats();
        let index = self
            .ctx
            .control
            .index_progress()?
            .into_iter()
            .map(|r| IndexProgress {
                repo: Some(RepoPath::new(&r.namespace, &r.name)),
                pack_id: r.pack_id,
                objects_done: r.objects_done,
                objects_total: r.objects_total,
                bytes_done: r.bytes_done,
                bytes_total: r.bytes_total,
                phase: r.phase.clone(),
                started_at_unix_ms: r.started_at_unix_ms,
                resumed: r.resumed,
                percent: r.percent(),
                reinflations: r.reinflations,
                pauses: r.pauses,
            })
            .collect();
        Ok(Response::new(StatsResponse {
            namespaces: counts.namespaces,
            repos: counts.repos,
            principals: counts.principals,
            keys: counts.keys,
            grants: counts.grants,
            event_subscribers: bus.subscribers,
            events_published: bus.published,
            events_dropped: bus.dropped,
            event_buffer_capacity: bus.capacity,
            index,
            tombstoned_repos: extras.tombstoned_repos,
            tombstoned_bytes: extras.tombstoned_bytes,
            replication_configured: extras.replication_configured,
            replication_synced: extras.replication_synced,
            replication_failed: extras.replication_failed,
            replication_enqueued: extras.replication_enqueued,
            replication_bus_gaps: extras.replication_bus_gaps,
            replication_refs_total: extras.replication_refs_total,
            replication_refs_unmirrored: extras.replication_refs_unmirrored,
            standbys: extras.standbys,
            tenants_paid: extras.tenants_paid,
            tenants_grace: extras.tenants_grace,
            tenants_suspended: extras.tenants_suspended,
        }))
    }

    async fn storage_info(
        &self,
        request: Request<StorageInfoRequest>,
    ) -> Result<Response<StorageInfoResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let path = opt_repo_path(request.into_inner().repo, "repo")?;
        // A whole-server storage report spans tenants; a single repo's does
        // not.
        let access = match &path {
            Some(path) => Access::repo(path),
            None => Access::Server,
        };
        self.ctx
            .require(&caller, access, GrantLevel::Read, "Admin.StorageInfo")?;
        let info = self.ctx.control.storage_info(path.as_ref())?;
        Ok(Response::new(info))
    }

    async fn verify_chain(
        &self,
        request: Request<VerifyChainRequest>,
    ) -> Result<Response<VerifyChainResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        let inner = request.into_inner();
        // The same split `StorageInfo` makes: a question about one repository
        // is that repository's business; a walk that starts at an artifact has
        // no repo until the build record names one, and the record is
        // server-wide evidence — so it takes a server-wide Read.
        let path = opt_repo_path(inner.repo.clone(), "repo")?;
        let access = match &path {
            Some(path) => Access::repo(path),
            None => Access::Server,
        };
        self.ctx
            .require(&caller, access, GrantLevel::Read, "Admin.VerifyChain")?;
        if path.is_none() && inner.artifact_sha256.is_empty() {
            return Err(Status::invalid_argument(
                "name a repository, or an artifact_sha256 to start from",
            ));
        }
        let chain = self.ctx.control.verify_chain(&inner)?;
        Ok(Response::new(chain))
    }

    /// **"If a tenant's cap rose to N bytes, could you serve it?"**
    ///
    /// Server-wide Read, the same as `Stats` and a repo-less `StorageInfo`:
    /// the answer is about the whole filesystem every tenant shares, so it is
    /// not one namespace's business even when a namespace is named — a
    /// namespace-scoped grant here would let one tenant read the box's free
    /// space by asking about itself.
    async fn capacity(
        &self,
        request: Request<CapacityRequest>,
    ) -> Result<Response<CapacityResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.ctx
            .require(&caller, Access::Server, GrantLevel::Read, "Admin.Capacity")?;
        let answer = self.ctx.control.capacity(&request.into_inner())?;
        Ok(Response::new(answer))
    }

    // ── the runtime data-set growth (`DATA-SET-GROWTH-FLOW.md` §3) ────────

    /// **Begin a growth.** gunnar-admin: it drains the box, unmounts `/data`
    /// and hands the members to an outside actor — the one verb here whose
    /// failure mode is minutes of refused pushes (T12), so it is held to the
    /// highest level and to a go-ahead the control plane verifies (T14).
    async fn grow_start(
        &self,
        request: Request<GrowStartRequest>,
    ) -> Result<Response<GrowStatusResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.ctx
            .require(&caller, Access::Server, GrantLevel::Admin, "Admin.GrowStart")?;
        let request = request.into_inner();
        if request.go_ahead.is_empty() {
            return Err(Status::invalid_argument(
                "`go_ahead` is required: the signed go-ahead the growth is started on (D3). \
                 `gunnar control grow start` mints an operator one",
            ));
        }
        Ok(Response::new(self.ctx.control.grow_start(&request)?))
    }

    /// The growth's phase, with a credential. The SAME answer is served
    /// without one on `Bootstrap.GrowStatus` (see there for why); this arm
    /// exists so a credentialled operator script does not have to know that.
    async fn grow_status(
        &self,
        request: Request<GrowStatusRequest>,
    ) -> Result<Response<GrowStatusResponse>, Status> {
        self.ctx
            .authorize(&request, Access::Server, GrantLevel::Read, "Admin.GrowStatus")?;
        Ok(Response::new(self.ctx.control.grow_status()?))
    }

    /// **Look again** — the outside actor says the members are back and
    /// bigger; or, with `give_up`, the operator says stop waiting and serve
    /// what is there. gunnar-admin, like `GrowStart`.
    async fn grow_resume(
        &self,
        request: Request<GrowResumeRequest>,
    ) -> Result<Response<GrowStatusResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.ctx
            .require(&caller, Access::Server, GrantLevel::Admin, "Admin.GrowResume")?;
        Ok(Response::new(
            self.ctx.control.grow_resume(request.into_inner().give_up)?,
        ))
    }

    /// **The tunnel address of a box past its bootstrap** (2026-09-21, T17).
    /// Server-wide Admin: the address decides what the hub routes to this
    /// box, the bar `GrowStart` takes for the same kind of reason. A READ
    /// (empty address) is held to the same bar — the unauthenticated read is
    /// `Bootstrap.AssignTunnelAddress` with an empty address.
    async fn set_tunnel_address(
        &self,
        request: Request<gunnar_proto::AssignTunnelAddressRequest>,
    ) -> Result<Response<gunnar_proto::AssignTunnelAddressResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.ctx
            .require(&caller, Access::Server, GrantLevel::Admin, "Admin.SetTunnelAddress")?;
        let address = request.into_inner().address;
        let answer = self.ctx.control.set_tunnel_address(address.trim())?;
        if answer.changed {
            tracing::warn!(target: crate::auth::AUDIT_TARGET, event = "grpc.call", method = "Admin.SetTunnelAddress", address = %answer.address, wg_public_key = %answer.wg_public_key, tunnel_up = answer.tunnel_up, decision = "allow", "an enrolled box was given its tunnel address: register its key on the hub at this address");
        }
        Ok(Response::new(answer))
    }

    /// **Which CA does this box mint with** (2026-09-17).
    ///
    /// # Why this verb exists at all
    ///
    /// `gunnar issuer show` answers this from the data root, and the appliance
    /// has no shell — that is the product's whole shape. The issuer
    /// certificate went out over exactly one RPC, `Bootstrap.Enroll`, and that
    /// window closes permanently on the first enrol. So on a RESTORED box, the
    /// one place the question actually bites, the certificate every client
    /// pins was unreadable from outside by any means at all. The first-start
    /// log line even tells the operator to run `gunnar issuer show --pem` on a
    /// machine where there is nothing to run it with.
    ///
    /// # The access level, and the one that was rejected
    ///
    /// **Server-wide `Read`** — the bar `Health`, `Stats` and `StorageInfo`
    /// take, and for their reason.
    ///
    /// The tempting argument is that a CA certificate is public by
    /// construction: every client holds a copy, it is presented in every
    /// handshake, and its secrecy protects nothing. True of the certificate —
    /// and this answer is more than the certificate. It enumerates the trusted
    /// CA list, says whether the key was minted here or IMPORTED (so whether a
    /// copy exists elsewhere), and dates the adoption: a rotation in progress
    /// is legible in it. That is reconnaissance of exactly the kind
    /// `Admin.Health` is authenticated for — it reports where the listeners
    /// are — and the same answer applies: a caller who cannot present any
    /// credential has no business enumerating this box's PKI state.
    ///
    /// So **`UNAUTHENTICATED_METHODS` was rejected**, and deliberately: that
    /// list has one reason, and it is that the identity table cannot answer
    /// during first boot or a data-set growth. Nothing here needs that. A
    /// restored box has a working identity table BY DEFINITION — restoring it
    /// is what the restore did — so the operator asking this question is an
    /// operator who can sign for it.
    ///
    /// **`Admin` was rejected too**, in the other direction. This is a read
    /// with no side effect on a document whose payload is public; putting it
    /// behind the level that mints principals would mean the person verifying
    /// a pin must hold the credential that can rewrite the ACL, which is a
    /// worse outcome than the one it guards against. `Read` at `Server` scope
    /// is the level for a server-wide fact, and this is one.
    async fn issuer_info(
        &self,
        request: Request<gunnar_proto::IssuerInfoRequest>,
    ) -> Result<Response<gunnar_proto::IssuerInfoResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.ctx
            .require(&caller, Access::Server, GrantLevel::Read, "Admin.IssuerInfo")?;
        // Read-only all the way down: the seam's one implementation opens
        // `ServerStores::pki_readonly()`, whose secret store refuses every
        // call. There is no branch here that could mint or rotate, and there
        // is no value in scope that could carry a private key.
        Ok(Response::new(self.ctx.control.issuer_info()?))
    }

    /// **`Admin.AuditTrail`** (lane T11): the audit chain's shape, so that "is
    /// the record still complete, and is it the same record" can be asked of a
    /// running box — in particular of a RESTORED one, which is the only place
    /// the question has ever really mattered.
    ///
    /// `Read` at `Server` scope, the level for a server-wide fact, and the
    /// same level as `Health` and `IssuerInfo`. It is deliberately not higher:
    /// the person checking that a restore kept the trail is exactly the person
    /// who must not have to hold a credential that can rewrite the ACL. It is
    /// deliberately not lower either — the response is a count and a hash and
    /// carries no entry, but who-audits-whom is still a server-wide fact.
    ///
    /// **This call is itself audited**, like every other authorised RPC, so
    /// the tip it reports is the tip as of the decision that let it read —
    /// which is what makes two boxes comparable at a named `at_seq` and not at
    /// a tip that the act of asking has already moved.
    async fn audit_trail(
        &self,
        request: Request<gunnar_proto::AuditTrailRequest>,
    ) -> Result<Response<gunnar_proto::AuditTrailResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.ctx
            .require(&caller, Access::Server, GrantLevel::Read, "Admin.AuditTrail")?;
        let at_seq = request.into_inner().at_seq;
        Ok(Response::new(self.ctx.control.audit_trail(at_seq)?))
    }

    /// **`Admin.Topology`** (2026-10-04): the estate's boxes, as this gunnar
    /// knows them — see [`crate::topology`], where every decision is.
    ///
    /// Server-wide `Read`, the level of `Health`, whose twin rows it reuses:
    /// the rows name addresses and principals, never a secret. A MACHINE that
    /// asks is marked in the caller log as one that draws the picture — a
    /// console — so the console that is asking is on the picture it gets.
    async fn topology(
        &self,
        request: Request<gunnar_proto::TopologyRequest>,
    ) -> Result<Response<gunnar_proto::TopologyResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.ctx.require(&caller, Access::Server, GrantLevel::Read, "Admin.Topology")?;
        let now = estate_clock::now_unix_ms();
        if caller.machine && caller.authenticated {
            let address = request.remote_addr().map(|a| a.to_string()).unwrap_or_default();
            self.ctx.callers.note(&caller.principal_id, &address, now, true);
        }
        let silent_after_ms = self.ctx.heartbeats.bounds().silent_after_ms;
        // A plane that keeps no replication counters still has a topology:
        // its standbys are simply none.
        let standbys = self.ctx.control.stats_extras().map(|e| e.standbys).unwrap_or_default();
        let mut facts = self.ctx.control.topology_facts();
        if facts.address.is_empty() {
            // Unpaired: where this listener is bound is where this box is.
            facts.address = self.ctx.config.bind.to_string();
        }
        let input = crate::topology::Input {
            facts,
            is_twin: self.ctx.control.is_twin(),
            twins: self.twin_rows(),
            primary: self.ctx.control.primary_heartbeat(silent_after_ms),
            standbys,
            callers: self.ctx.callers.sightings(),
            now_ms: now,
            silent_after_ms,
        };
        Ok(Response::new(crate::topology::assemble(&input)))
    }

    /// **Sign a twin's replication certificate** (lane T9).
    ///
    /// Server-wide `Admin`, deliberately the level that mints principals and
    /// not the `Read` its neighbour `IssuerInfo` takes: this verb puts a
    /// credential into the stream that carries every sealed secret on the box.
    /// The seam adds the two refusals that are about THIS box rather than the
    /// caller — no mutual replication listener, principal not in group `twin`.
    async fn issue_twin_certificate(
        &self,
        request: Request<gunnar_proto::IssueTwinCertificateRequest>,
    ) -> Result<Response<gunnar_proto::IssueTwinCertificateResponse>, Status> {
        let caller = self.ctx.caller(&request)?;
        self.ctx
            .require(&caller, Access::Server, GrantLevel::Admin, "Admin.IssueTwinCertificate")?;
        let r = request.into_inner();
        let (cert_pem, issuer_pem, port, host) = self.ctx.control.issue_twin_certificate(&r.principal, &r.csr_pem)?;
        tracing::warn!(
            target: crate::auth::AUDIT_TARGET,
            event = "grpc.call",
            method = "Admin.IssueTwinCertificate",
            principal = %caller.principal_id,
            twin = %r.principal,
            replication_port = port,
            replication_host = %host,
            decision = "allow",
            "a twin replication certificate was issued and registered"
        );
        Ok(Response::new(gunnar_proto::IssueTwinCertificateResponse {
            cert_pem,
            issuer_pem,
            replication_port: u32::from(port),
            replication_host: host,
        }))
    }
}

/// A clock-sync report as it goes on the wire. One conversion, used by both
/// `Admin.Health` and `Bootstrap.Status`, so the two can never describe the
/// same attempt differently.
pub(crate) fn clock_sync(report: estate_clock::Report) -> gunnar_proto::ClockSync {
    gunnar_proto::ClockSync {
        trigger: report.trigger.as_str().to_owned(),
        at_unix_ms: report.at_unix_ms,
        outcome: report.outcome.as_str().to_owned(),
        offset_ms: report.offset_ms,
        detail: report.detail,
        sources: report
            .sources
            .into_iter()
            .map(|row| gunnar_proto::ClockSource {
                source: row.source,
                answered: row.offset_ms.is_some() && row.error.is_none(),
                offset_ms: row.offset_ms.unwrap_or(0),
                error: row.error.unwrap_or_default(),
            })
            .collect(),
    }
}

impl AdminService {
    /// The primary's twin rows: every member of group `twin` plus every
    /// caller that reported as a twin. A group that cannot be listed (a plane
    /// with no groups, or no `twin` group yet) contributes no names — the
    /// callers that polled still show.
    fn twin_rows(&self) -> Vec<gunnar_proto::TwinHeartbeat> {
        let members: Vec<String> = self
            .ctx
            .control
            .list_group_members(crate::TWIN_GROUP)
            .map(|ps| ps.into_iter().map(|p| p.id.to_string()).collect())
            .unwrap_or_default();
        self.ctx.heartbeats.view(&members, estate_clock::now_unix_ms())
    }
}

#[cfg(test)]
mod clock_tests {
    use super::*;
    use estate_clock::report::{Outcome, Trigger, Report, SourceReport};

    #[test]
    fn a_failed_source_goes_on_the_wire_as_not_answered_with_its_reason() {
        let wire = clock_sync(Report {
            trigger: Trigger::Boot,
            at_unix_ms: 1_789_394_521_628,
            outcome: Outcome::NoQuorum,
            offset_ms: 0,
            detail: "clock unchanged".into(),
            sources: vec![
                SourceReport {
                    source: "peer://10.13.8.99:50123".into(),
                    offset_ms: None,
                    error: Some("did not accept a connection".into()),
                },
                SourceReport {
                    source: "https://api.stripe.com/".into(),
                    offset_ms: Some(-7_199_500),
                    error: None,
                },
            ],
        });
        assert_eq!((wire.trigger.as_str(), wire.outcome.as_str()), ("boot", "no-quorum"));
        assert!(!wire.sources[0].answered);
        assert_eq!(wire.sources[0].error, "did not accept a connection");
        assert!(wire.sources[1].answered);
        assert_eq!(wire.sources[1].offset_ms, -7_199_500);
    }
}