canton_admin/

client.rs

//! Ledger-API admin client: party management + user self-inspect.

use std::sync::Arc;

use canton_core::auth::{self, Intercepted};
use canton_core::telemetry::{self, TRANSPORT_GRPC};
use canton_core::{Config, Error, Result};
use canton_proto::com::daml::ledger::api::v2 as lapi;
use canton_proto::com::daml::ledger::api::v2::admin as pb;
use lapi::package_service_client::PackageServiceClient;
use pb::party_management_service_client::PartyManagementServiceClient;
use pb::user_management_service_client::UserManagementServiceClient;
use tonic::transport::Channel;

/// A client for Canton's Ledger-API **admin** services: `PartyManagementService`
/// and `UserManagementService`.
///
/// Served on the Ledger API port (e.g. `:3901`) — the same endpoint and auth as
/// [`canton_ledger`]. Party-management RPCs require the `ParticipantAdmin`
/// right; the `current_user*` self-inspect calls do not.
///
/// [`canton_ledger`]: https://docs.rs/canton-ledger
#[derive(Clone, Debug)]
pub struct AdminClient {
    channel: Channel,
    config: Arc<Config>,
}

/// Build a gRPC service client on the authenticated channel, with this
/// client's decode limit applied.
///
/// A macro rather than a function because `max_decoding_message_size` is an
/// inherent method on each generated client — tonic exposes no trait for it —
/// so there is nothing to be generic over. Keeping every construction site
/// behind one expansion is the point: `tonic`'s 4 MiB default is small enough
/// that a real ACS page trips it, and a new RPC added later would otherwise
/// pick the default up silently.
macro_rules! service {
    ($self:ident, $ctor:expr) => {
        $ctor($self.intercepted().await?)
            .max_decoding_message_size($self.config.max_decoding_message_size())
    };
}

impl AdminClient {
    /// Build a lazily-connected client. Returns immediately; the TCP/TLS
    /// handshake happens on the first RPC.
    ///
    /// # Errors
    /// Returns [`Error::InvalidRequest`] if the endpoint is not a valid URI.
    pub fn connect_lazy(config: Config) -> Result<Self> {
        Ok(Self {
            channel: config.connect_channel()?,
            config: Arc::new(config),
        })
    }

    /// A channel wrapped with a fresh bearer-token interceptor for this call.
    async fn intercepted(&self) -> Result<Intercepted> {
        auth::intercepted(&self.channel, self.config.auth()).await
    }

    /// Run a unary operation under the configured retry policy (no-op when
    /// retrying is not enabled).
    async fn with_retry<T, F, Fut>(&self, op: F) -> Result<T>
    where
        F: FnMut() -> Fut,
        Fut: std::future::Future<Output = Result<T>>,
    {
        canton_core::retry::run_with_retry(self.config.retry(), op).await
    }

    // ---- PartyManagementService --------------------------------------------

    /// The participant's id (`GetParticipantId`). Works with any authenticated
    /// token.
    ///
    /// # Errors
    /// Returns an [`Error`] if the RPC fails.
    pub async fn participant_id(&self) -> Result<String> {
        telemetry::instrument("participant_id", TRANSPORT_GRPC, async {
            self.with_retry(|| async {
                let mut client = service!(self, PartyManagementServiceClient::new);
                Ok(client
                    .get_participant_id(pb::GetParticipantIdRequest {})
                    .await?
                    .into_inner()
                    .participant_id)
            })
            .await
        })
        .await
    }

    /// Allocate a new party (`AllocateParty`). `party_id_hint` suggests the
    /// local part of the party id; the participant returns the actual
    /// [`PartyDetails`](pb::PartyDetails) (`party = "<hint>::<fingerprint>"`).
    ///
    /// Requires the `ParticipantAdmin` right. Not retried: party allocation is
    /// a non-idempotent topology mutation (a lost response must not double
    /// allocate), so a transient failure surfaces to the caller.
    ///
    /// # Errors
    /// Returns an [`Error`] if authentication/authorization fails, the RPC
    /// fails, or the response carries no party details.
    pub async fn allocate_party(&self, party_id_hint: Option<&str>) -> Result<pb::PartyDetails> {
        let party_id_hint = party_id_hint.unwrap_or_default().to_string();
        telemetry::instrument("allocate_party", TRANSPORT_GRPC, async move {
            let mut client = service!(self, PartyManagementServiceClient::new);
            let response = client
                .allocate_party(pb::AllocatePartyRequest {
                    party_id_hint,
                    ..Default::default()
                })
                .await?
                .into_inner();
            response.party_details.ok_or_else(|| {
                Error::UnexpectedResponse("allocate_party returned no party details".to_string())
            })
        })
        .await
    }

    /// Ask the participant what allocating an external party for `public_key`
    /// would look like, without allocating it.
    ///
    /// The first half of external-party onboarding. The participant computes
    /// the key's fingerprint and the party id derived from it, builds the
    /// onboarding topology transactions, and returns one hash over them. None
    /// of it is submitted: that is [`allocate_external_party`], after the key
    /// has signed the hash.
    ///
    /// [`allocate_external_party`]: Self::allocate_external_party
    ///
    /// # Errors
    /// Returns an [`Error`] if the RPC fails, or [`Error::UnexpectedResponse`]
    /// if the participant answers without a hash to sign or without the
    /// fingerprint — either of which would leave the caller unable to sign as
    /// this party at all.
    pub async fn generate_external_party_topology(
        &self,
        synchronizer: &str,
        party_hint: &str,
        public_key: &canton_signer::PublicKey,
    ) -> Result<crate::external_party::ExternalPartyTopology> {
        let request = pb::GenerateExternalPartyTopologyRequest {
            synchronizer: synchronizer.to_string(),
            party_hint: party_hint.to_string(),
            public_key: Some(public_key.into()),
            ..Default::default()
        };
        telemetry::instrument(
            "generate_external_party_topology",
            TRANSPORT_GRPC,
            async move {
                let response = self
                    .with_retry(|| {
                        let request = request.clone();
                        async move {
                            let mut client = service!(self, PartyManagementServiceClient::new);
                            Ok(client
                                .generate_external_party_topology(request)
                                .await?
                                .into_inner())
                        }
                    })
                    .await?;
                crate::external_party::ExternalPartyTopology::from_response(response, public_key)
            },
        )
        .await
    }

    /// Allocate the external party described by `topology`, authorized by
    /// `signer`.
    ///
    /// The second half. The signature is over
    /// [`multi_hash`](crate::external_party::ExternalPartyTopology::multi_hash)
    /// — one hash covering every onboarding transaction, so one signature
    /// authorizes the whole set — and it is what proves the party controls the
    /// key being registered for it.
    ///
    /// `user_id` gets `act_as` rights on the new party; empty grants none.
    ///
    /// Returns the allocated party id. This waits for the allocation to reach
    /// the synchronizer, so the party is usable when it returns.
    ///
    /// # Errors
    /// Returns an [`Error`] if signing fails or the RPC does. A signature made
    /// by a key other than the one in the topology is rejected by the
    /// participant, not here.
    pub async fn allocate_external_party(
        &self,
        synchronizer: &str,
        topology: &crate::external_party::ExternalPartyTopology,
        signer: &dyn canton_signer::Signer,
        user_id: &str,
    ) -> Result<String> {
        let signature = signer.sign(topology.multi_hash()).await?;
        let request = pb::AllocateExternalPartyRequest {
            synchronizer: synchronizer.to_string(),
            onboarding_transactions: topology
                .transactions()
                .iter()
                .map(
                    |transaction| pb::allocate_external_party_request::SignedTransaction {
                        transaction: transaction.clone(),
                        // The multi-hash signature below covers every transaction,
                        // so none needs one of its own.
                        signatures: Vec::new(),
                    },
                )
                .collect(),
            multi_hash_signatures: vec![(&signature).into()],
            identity_provider_id: String::new(),
            // The party is not usable until it is on the synchronizer, and a
            // caller that just allocated it is about to use it.
            wait_for_allocation: Some(true),
            user_id: user_id.to_string(),
        };
        telemetry::instrument("allocate_external_party", TRANSPORT_GRPC, async move {
            // Deliberately not retried, exactly as `allocate_party` is not.
            // `AllocateExternalPartyRequest` carries no command or submission
            // id, so there is no idempotency key that would make a second
            // attempt a duplicate — and `wait_for_allocation` makes the call
            // long enough that losing the response to a committed allocation is
            // realistic. Retrying would be a guess about Canton's behaviour
            // dressed up as resilience.
            let response = {
                let mut client = service!(self, PartyManagementServiceClient::new);
                client.allocate_external_party(request).await?.into_inner()
            };
            if response.party_id != topology.party_id() {
                return Err(Error::UnexpectedResponse(format!(
                    "allocate_external_party allocated `{}`, not the `{}` that was signed for",
                    response.party_id,
                    topology.party_id()
                )));
            }
            if response.party_id.is_empty() {
                return Err(Error::UnexpectedResponse(
                    "allocate_external_party returned no party id".to_string(),
                ));
            }
            Ok(response.party_id)
        })
        .await
    }

    /// List one page of known parties (`ListKnownParties`). Pass the returned
    /// token to fetch the next page; `page_size` `0` uses the server default.
    ///
    /// # Errors
    /// Returns an [`Error`] if authentication/authorization or the RPC fails.
    pub async fn list_known_parties_page(
        &self,
        page_size: i32,
        page_token: Option<String>,
    ) -> Result<(Vec<pb::PartyDetails>, Option<String>)> {
        telemetry::instrument("list_known_parties_page", TRANSPORT_GRPC, async move {
            self.with_retry(|| {
                let page_token = page_token.clone().unwrap_or_default();
                async move {
                    let mut client = service!(self, PartyManagementServiceClient::new);
                    let response = client
                        .list_known_parties(pb::ListKnownPartiesRequest {
                            page_token,
                            page_size,
                            ..Default::default()
                        })
                        .await?
                        .into_inner();
                    let next =
                        (!response.next_page_token.is_empty()).then_some(response.next_page_token);
                    Ok((response.party_details, next))
                }
            })
            .await
        })
        .await
    }

    /// All known parties, following pagination internally (`ListKnownParties`).
    ///
    /// # Errors
    /// Returns an [`Error`] if authentication/authorization or any page RPC
    /// fails.
    pub async fn list_known_parties(&self) -> Result<Vec<pb::PartyDetails>> {
        telemetry::instrument("list_known_parties", TRANSPORT_GRPC, async {
            let mut all = Vec::new();
            let mut page_token = String::new();
            let mut seen = std::collections::HashSet::new();
            for page in 1.. {
                let sent = page_token.clone();
                let (parties, next) = self
                    .with_retry(|| {
                        let page_token = sent.clone();
                        async move {
                            let mut client = service!(self, PartyManagementServiceClient::new);
                            let response = client
                                .list_known_parties(pb::ListKnownPartiesRequest {
                                    page_token,
                                    ..Default::default()
                                })
                                .await?
                                .into_inner();
                            Ok((response.party_details, response.next_page_token))
                        }
                    })
                    .await?;
                all.extend(parties);
                if next.is_empty() {
                    break;
                }
                // A server that never advances the token, cycles between a
                // few, or mints a fresh one forever would have this loop run
                // until memory ran out. Stopping quietly is no better: the
                // caller receives a prefix of the party list with nothing to
                // say it is a prefix, and "which parties exist" is a question
                // whose wrong answer looks exactly like a right one. So a
                // token seen before, or more pages than any participant has,
                // fails.
                if next == sent || !seen.insert(next.clone()) {
                    return Err(Error::UnexpectedResponse(format!(
                        "the participant repeated a page token after {} parties; \
                         the list is incomplete and cannot be continued",
                        all.len()
                    )));
                }
                if page >= canton_core::MAX_LIST_PAGES {
                    return Err(Error::UnexpectedResponse(format!(
                        "the party listing did not end after {} pages and {} parties; \
                         the list is incomplete and cannot be continued",
                        canton_core::MAX_LIST_PAGES,
                        all.len()
                    )));
                }
                page_token = next;
            }
            Ok(all)
        })
        .await
    }

    /// Details for specific parties (`GetParties`). Unknown parties are silently
    /// omitted from the result rather than erroring.
    ///
    /// # Errors
    /// Returns an [`Error`] if authentication/authorization or the RPC fails.
    pub async fn get_parties(&self, parties: Vec<String>) -> Result<Vec<pb::PartyDetails>> {
        telemetry::instrument("get_parties", TRANSPORT_GRPC, async move {
            self.with_retry(|| {
                let parties = parties.clone();
                async move {
                    let mut client = service!(self, PartyManagementServiceClient::new);
                    Ok(client
                        .get_parties(pb::GetPartiesRequest {
                            parties,
                            ..Default::default()
                        })
                        .await?
                        .into_inner()
                        .party_details)
                }
            })
            .await
        })
        .await
    }

    // ---- UserManagementService (self-inspect) ------------------------------

    /// The authenticated user's own record (`GetUser` with an empty user id).
    /// Authorized for any authenticated token (no admin right needed).
    ///
    /// # Errors
    /// Returns an [`Error`] if the RPC fails or the response carries no user.
    pub async fn current_user(&self) -> Result<pb::User> {
        self.get_user("").await
    }

    /// A user's record by id (`GetUser`); an empty `user_id` means the
    /// authenticated user.
    ///
    /// # Errors
    /// Returns an [`Error`] if the RPC fails or the response carries no user.
    pub async fn get_user(&self, user_id: &str) -> Result<pb::User> {
        let user_id = user_id.to_string();
        telemetry::instrument("get_user", TRANSPORT_GRPC, async move {
            self.with_retry(|| {
                let user_id = user_id.clone();
                async move {
                    let mut client = service!(self, UserManagementServiceClient::new);
                    let response = client
                        .get_user(pb::GetUserRequest {
                            user_id,
                            ..Default::default()
                        })
                        .await?
                        .into_inner();
                    response.user.ok_or_else(|| {
                        Error::UnexpectedResponse("get_user returned no user".to_string())
                    })
                }
            })
            .await
        })
        .await
    }

    /// The authenticated user's own rights (`ListUserRights` with an empty user
    /// id). Authorized for any authenticated token.
    ///
    /// # Errors
    /// Returns an [`Error`] if the RPC fails.
    pub async fn current_user_rights(&self) -> Result<Vec<pb::Right>> {
        self.list_user_rights("").await
    }

    /// A user's rights by id (`ListUserRights`); an empty `user_id` means the
    /// authenticated user.
    ///
    /// # Errors
    /// Returns an [`Error`] if the RPC fails.
    pub async fn list_user_rights(&self, user_id: &str) -> Result<Vec<pb::Right>> {
        let user_id = user_id.to_string();
        telemetry::instrument("list_user_rights", TRANSPORT_GRPC, async move {
            self.with_retry(|| {
                let user_id = user_id.clone();
                async move {
                    let mut client = service!(self, UserManagementServiceClient::new);
                    Ok(client
                        .list_user_rights(pb::ListUserRightsRequest {
                            user_id,
                            ..Default::default()
                        })
                        .await?
                        .into_inner()
                        .rights)
                }
            })
            .await
        })
        .await
    }

    // ---- PackageService (read) ----------------------------------------------

    /// The ids of all Daml packages known to the participant
    /// (`PackageService.ListPackages`).
    ///
    /// # Errors
    /// Returns an [`Error`] if authentication or the RPC fails.
    pub async fn list_packages(&self) -> Result<Vec<String>> {
        telemetry::instrument("list_packages", TRANSPORT_GRPC, async {
            self.with_retry(|| async {
                let mut client = service!(self, PackageServiceClient::new);
                Ok(client
                    .list_packages(lapi::ListPackagesRequest {})
                    .await?
                    .into_inner()
                    .package_ids)
            })
            .await
        })
        .await
    }

    /// The status of a package on the participant
    /// (`PackageService.GetPackageStatus`): `Registered` if the package is
    /// known, `Unspecified` otherwise.
    ///
    /// # Errors
    /// Download a package the participant knows, as `ArchivePayload` bytes.
    ///
    /// This is what `canton_lf::decode_payload` reads — the *inner* half of an
    /// archive. A DAR entry is the whole `Archive`, payload and hash together,
    /// and `canton_lf::decode_package` reads that one; the Ledger API returns
    /// the parts separately, so the two are different bytes and each has its
    /// own decoder.
    ///
    /// Bindings can therefore be generated from what a participant has actually
    /// vetted, rather than from a DAR file that has to be found somewhere — and
    /// for a package that only ever existed on a network, there may be no such
    /// file at all.
    ///
    /// The payload is hashed and the digest checked against `package_id` before the bytes are
    /// handed back. A package id **is** the hash of its payload, so that check
    /// is what makes asking by id pin the content exactly, with nothing further
    /// to checksum.
    ///
    /// # Errors
    /// Returns an [`Error`] if the RPC fails, and
    /// [`Error::UnexpectedResponse`] if the participant answers with no payload
    /// or whose bytes do not hash to the id that was asked for.
    pub async fn get_package(&self, package_id: &str) -> Result<Vec<u8>> {
        let package_id = package_id.to_string();
        let package_for_check = package_id.clone();
        telemetry::instrument("get_package", TRANSPORT_GRPC, async move {
            let response = self
                .with_retry(|| {
                    let package_id = package_id.clone();
                    async move {
                        let mut client = service!(self, PackageServiceClient::new);
                        Ok(client
                            .get_package(lapi::GetPackageRequest { package_id })
                            .await?
                            .into_inner())
                    }
                })
                .await?;
            if response.archive_payload.is_empty() {
                return Err(Error::UnexpectedResponse(
                    "the participant returned an empty archive".to_string(),
                ));
            }
            // A package id *is* the SHA-256 of its payload, so the check is to
            // hash the bytes — not to read the `hash` field back. That field is
            // the server's own claim about what it sent, so comparing it to the
            // id proves only that the server is self-consistent, and an empty
            // one skipped the check altogether. Hashing the payload is what
            // makes asking by id pin the content.
            let computed = hex_sha256(&response.archive_payload);
            if computed != package_for_check {
                return Err(Error::UnexpectedResponse(format!(
                    "asked for package {package_for_check} and got {} bytes hashing to {computed}",
                    response.archive_payload.len()
                )));
            }
            Ok(response.archive_payload)
        })
        .await
    }

    /// Returns an [`Error`] if authentication or the RPC fails.
    pub async fn get_package_status(&self, package_id: &str) -> Result<lapi::PackageStatus> {
        let package_id = package_id.to_string();
        telemetry::instrument("get_package_status", TRANSPORT_GRPC, async move {
            self.with_retry(|| {
                let package_id = package_id.clone();
                async move {
                    let mut client = service!(self, PackageServiceClient::new);
                    let response = client
                        .get_package_status(lapi::GetPackageStatusRequest { package_id })
                        .await?
                        .into_inner();
                    Ok(response.package_status())
                }
            })
            .await
        })
        .await
    }

    /// The specific parties the authenticated user may act as — the `CanActAs`
    /// subset of [`Self::current_user_rights`]. Supersedes the ad-hoc party
    /// discovery used in early tests.
    ///
    /// Note: a user granted the wildcard `CanActAsAnyParty` right may act as
    /// *any* party yet carries no explicit `CanActAs` entries, so this returns
    /// an empty list for such users — an empty result means "no specific
    /// parties", not necessarily "cannot act". Inspect
    /// [`Self::current_user_rights`] directly if you need the wildcard.
    ///
    /// # Errors
    /// Returns an [`Error`] if the underlying rights RPC fails.
    pub async fn acting_parties(&self) -> Result<Vec<String>> {
        let rights = self.current_user_rights().await?;
        Ok(rights
            .into_iter()
            .filter_map(|right| match right.kind {
                Some(pb::right::Kind::CanActAs(can)) => Some(can.party),
                _ => None,
            })
            .collect())
    }
}

/// The lowercase hex SHA-256 of `bytes` — which, for a Daml-LF package
/// payload, *is* its package id.
fn hex_sha256(bytes: &[u8]) -> String {
    use sha2::{Digest as _, Sha256};
    Sha256::digest(bytes)
        .iter()
        .fold(String::with_capacity(64), |mut out, byte| {
            use std::fmt::Write as _;
            let _ = write!(out, "{byte:02x}");
            out
        })
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod package_integrity_tests {
    use super::hex_sha256;

    /// The known-answer test, so the helper is pinned to SHA-256 rather than to
    /// whatever it currently computes.
    #[test]
    fn the_digest_is_sha256_in_lowercase_hex() {
        assert_eq!(
            hex_sha256(b""),
            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
        );
        assert_eq!(
            hex_sha256(b"abc"),
            "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"
        );
    }

    /// The property `get_package` relies on: the committed V2 corpus is named
    /// by the id its bytes hash to, and that is what makes asking by id pin the
    /// content. If this helper ever stopped being SHA-256, those names would
    /// stop meaning anything and nothing else would notice.
    #[test]
    fn a_committed_payload_hashes_to_the_id_in_its_name() {
        let dir = concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../../testdata/token-standard-v2"
        );
        let Ok(entries) = std::fs::read_dir(dir) else {
            return; // not checked out — the drift guard covers this corpus too
        };
        let mut checked = 0;
        for entry in entries.flatten() {
            let name = entry.file_name().to_string_lossy().to_string();
            let Some(stem) = name.strip_suffix(".lfpayload") else {
                continue;
            };
            let Some((_, id)) = stem.rsplit_once('-') else {
                continue;
            };
            let bytes = std::fs::read(entry.path()).expect("the payload reads");
            assert_eq!(hex_sha256(&bytes), id, "{name} does not hash to its own id");
            checked += 1;
        }
        assert!(checked >= 9, "expected the V2 corpus, checked {checked}");
    }
}