canton_admin/client.rs
1//! Ledger-API admin client: party management + user self-inspect.
2
3use std::sync::Arc;
4
5use canton_core::auth::{self, Intercepted};
6use canton_core::telemetry::{self, TRANSPORT_GRPC};
7use canton_core::{Config, Error, Result};
8use canton_proto::com::daml::ledger::api::v2 as lapi;
9use canton_proto::com::daml::ledger::api::v2::admin as pb;
10use lapi::package_service_client::PackageServiceClient;
11use pb::party_management_service_client::PartyManagementServiceClient;
12use pb::user_management_service_client::UserManagementServiceClient;
13use tonic::transport::Channel;
14
15/// A client for Canton's Ledger-API **admin** services: `PartyManagementService`
16/// and `UserManagementService`.
17///
18/// Served on the Ledger API port (e.g. `:3901`) — the same endpoint and auth as
19/// [`canton_ledger`]. Party-management RPCs require the `ParticipantAdmin`
20/// right; the `current_user*` self-inspect calls do not.
21///
22/// [`canton_ledger`]: https://docs.rs/canton-ledger
23#[derive(Clone, Debug)]
24pub struct AdminClient {
25 channel: Channel,
26 config: Arc<Config>,
27}
28
29impl AdminClient {
30 /// Build a lazily-connected client. Returns immediately; the TCP/TLS
31 /// handshake happens on the first RPC.
32 ///
33 /// # Errors
34 /// Returns [`Error::InvalidRequest`] if the endpoint is not a valid URI.
35 pub fn connect_lazy(config: Config) -> Result<Self> {
36 Ok(Self {
37 channel: config.connect_channel()?,
38 config: Arc::new(config),
39 })
40 }
41
42 /// A channel wrapped with a fresh bearer-token interceptor for this call.
43 async fn intercepted(&self) -> Result<Intercepted> {
44 auth::intercepted(&self.channel, self.config.auth()).await
45 }
46
47 /// Run a unary operation under the configured retry policy (no-op when
48 /// retrying is not enabled).
49 async fn with_retry<T, F, Fut>(&self, op: F) -> Result<T>
50 where
51 F: FnMut() -> Fut,
52 Fut: std::future::Future<Output = Result<T>>,
53 {
54 canton_core::retry::run_with_retry(self.config.retry(), op).await
55 }
56
57 // ---- PartyManagementService --------------------------------------------
58
59 /// The participant's id (`GetParticipantId`). Works with any authenticated
60 /// token.
61 ///
62 /// # Errors
63 /// Returns an [`Error`] if the RPC fails.
64 pub async fn participant_id(&self) -> Result<String> {
65 telemetry::instrument("participant_id", TRANSPORT_GRPC, async {
66 self.with_retry(|| async {
67 let mut client = PartyManagementServiceClient::new(self.intercepted().await?);
68 Ok(client
69 .get_participant_id(pb::GetParticipantIdRequest {})
70 .await?
71 .into_inner()
72 .participant_id)
73 })
74 .await
75 })
76 .await
77 }
78
79 /// Allocate a new party (`AllocateParty`). `party_id_hint` suggests the
80 /// local part of the party id; the participant returns the actual
81 /// [`PartyDetails`](pb::PartyDetails) (`party = "<hint>::<fingerprint>"`).
82 ///
83 /// Requires the `ParticipantAdmin` right. Not retried: party allocation is
84 /// a non-idempotent topology mutation (a lost response must not double
85 /// allocate), so a transient failure surfaces to the caller.
86 ///
87 /// # Errors
88 /// Returns an [`Error`] if authentication/authorization fails, the RPC
89 /// fails, or the response carries no party details.
90 pub async fn allocate_party(&self, party_id_hint: Option<&str>) -> Result<pb::PartyDetails> {
91 let party_id_hint = party_id_hint.unwrap_or_default().to_string();
92 telemetry::instrument("allocate_party", TRANSPORT_GRPC, async move {
93 let mut client = PartyManagementServiceClient::new(self.intercepted().await?);
94 let response = client
95 .allocate_party(pb::AllocatePartyRequest {
96 party_id_hint,
97 ..Default::default()
98 })
99 .await?
100 .into_inner();
101 response.party_details.ok_or_else(|| {
102 Error::UnexpectedResponse("allocate_party returned no party details".to_string())
103 })
104 })
105 .await
106 }
107
108 /// List one page of known parties (`ListKnownParties`). Pass the returned
109 /// token to fetch the next page; `page_size` `0` uses the server default.
110 ///
111 /// # Errors
112 /// Returns an [`Error`] if authentication/authorization or the RPC fails.
113 pub async fn list_known_parties_page(
114 &self,
115 page_size: i32,
116 page_token: Option<String>,
117 ) -> Result<(Vec<pb::PartyDetails>, Option<String>)> {
118 telemetry::instrument("list_known_parties_page", TRANSPORT_GRPC, async move {
119 self.with_retry(|| {
120 let page_token = page_token.clone().unwrap_or_default();
121 async move {
122 let mut client = PartyManagementServiceClient::new(self.intercepted().await?);
123 let response = client
124 .list_known_parties(pb::ListKnownPartiesRequest {
125 page_token,
126 page_size,
127 ..Default::default()
128 })
129 .await?
130 .into_inner();
131 let next =
132 (!response.next_page_token.is_empty()).then_some(response.next_page_token);
133 Ok((response.party_details, next))
134 }
135 })
136 .await
137 })
138 .await
139 }
140
141 /// All known parties, following pagination internally (`ListKnownParties`).
142 ///
143 /// # Errors
144 /// Returns an [`Error`] if authentication/authorization or any page RPC
145 /// fails.
146 pub async fn list_known_parties(&self) -> Result<Vec<pb::PartyDetails>> {
147 telemetry::instrument("list_known_parties", TRANSPORT_GRPC, async {
148 let mut all = Vec::new();
149 let mut page_token = String::new();
150 loop {
151 let sent = page_token.clone();
152 let (parties, next) = self
153 .with_retry(|| {
154 let page_token = sent.clone();
155 async move {
156 let mut client =
157 PartyManagementServiceClient::new(self.intercepted().await?);
158 let response = client
159 .list_known_parties(pb::ListKnownPartiesRequest {
160 page_token,
161 ..Default::default()
162 })
163 .await?
164 .into_inner();
165 Ok((response.party_details, response.next_page_token))
166 }
167 })
168 .await?;
169 all.extend(parties);
170 if next.is_empty() {
171 break;
172 }
173 // Guard against a server that never advances the token: without
174 // this a degenerate/buggy participant would loop forever.
175 if next == sent {
176 tracing::warn!(
177 "list_known_parties: server returned an unchanged page token; \
178 stopping pagination with {} parties collected",
179 all.len()
180 );
181 break;
182 }
183 page_token = next;
184 }
185 Ok(all)
186 })
187 .await
188 }
189
190 /// Details for specific parties (`GetParties`). Unknown parties are silently
191 /// omitted from the result rather than erroring.
192 ///
193 /// # Errors
194 /// Returns an [`Error`] if authentication/authorization or the RPC fails.
195 pub async fn get_parties(&self, parties: Vec<String>) -> Result<Vec<pb::PartyDetails>> {
196 telemetry::instrument("get_parties", TRANSPORT_GRPC, async move {
197 self.with_retry(|| {
198 let parties = parties.clone();
199 async move {
200 let mut client = PartyManagementServiceClient::new(self.intercepted().await?);
201 Ok(client
202 .get_parties(pb::GetPartiesRequest {
203 parties,
204 ..Default::default()
205 })
206 .await?
207 .into_inner()
208 .party_details)
209 }
210 })
211 .await
212 })
213 .await
214 }
215
216 // ---- UserManagementService (self-inspect) ------------------------------
217
218 /// The authenticated user's own record (`GetUser` with an empty user id).
219 /// Authorized for any authenticated token (no admin right needed).
220 ///
221 /// # Errors
222 /// Returns an [`Error`] if the RPC fails or the response carries no user.
223 pub async fn current_user(&self) -> Result<pb::User> {
224 self.get_user("").await
225 }
226
227 /// A user's record by id (`GetUser`); an empty `user_id` means the
228 /// authenticated user.
229 ///
230 /// # Errors
231 /// Returns an [`Error`] if the RPC fails or the response carries no user.
232 pub async fn get_user(&self, user_id: &str) -> Result<pb::User> {
233 let user_id = user_id.to_string();
234 telemetry::instrument("get_user", TRANSPORT_GRPC, async move {
235 self.with_retry(|| {
236 let user_id = user_id.clone();
237 async move {
238 let mut client = UserManagementServiceClient::new(self.intercepted().await?);
239 let response = client
240 .get_user(pb::GetUserRequest {
241 user_id,
242 ..Default::default()
243 })
244 .await?
245 .into_inner();
246 response.user.ok_or_else(|| {
247 Error::UnexpectedResponse("get_user returned no user".to_string())
248 })
249 }
250 })
251 .await
252 })
253 .await
254 }
255
256 /// The authenticated user's own rights (`ListUserRights` with an empty user
257 /// id). Authorized for any authenticated token.
258 ///
259 /// # Errors
260 /// Returns an [`Error`] if the RPC fails.
261 pub async fn current_user_rights(&self) -> Result<Vec<pb::Right>> {
262 self.list_user_rights("").await
263 }
264
265 /// A user's rights by id (`ListUserRights`); an empty `user_id` means the
266 /// authenticated user.
267 ///
268 /// # Errors
269 /// Returns an [`Error`] if the RPC fails.
270 pub async fn list_user_rights(&self, user_id: &str) -> Result<Vec<pb::Right>> {
271 let user_id = user_id.to_string();
272 telemetry::instrument("list_user_rights", TRANSPORT_GRPC, async move {
273 self.with_retry(|| {
274 let user_id = user_id.clone();
275 async move {
276 let mut client = UserManagementServiceClient::new(self.intercepted().await?);
277 Ok(client
278 .list_user_rights(pb::ListUserRightsRequest {
279 user_id,
280 ..Default::default()
281 })
282 .await?
283 .into_inner()
284 .rights)
285 }
286 })
287 .await
288 })
289 .await
290 }
291
292 // ---- PackageService (read) ----------------------------------------------
293
294 /// The ids of all Daml packages known to the participant
295 /// (`PackageService.ListPackages`).
296 ///
297 /// # Errors
298 /// Returns an [`Error`] if authentication or the RPC fails.
299 pub async fn list_packages(&self) -> Result<Vec<String>> {
300 telemetry::instrument("list_packages", TRANSPORT_GRPC, async {
301 self.with_retry(|| async {
302 let mut client = PackageServiceClient::new(self.intercepted().await?);
303 Ok(client
304 .list_packages(lapi::ListPackagesRequest {})
305 .await?
306 .into_inner()
307 .package_ids)
308 })
309 .await
310 })
311 .await
312 }
313
314 /// The status of a package on the participant
315 /// (`PackageService.GetPackageStatus`): `Registered` if the package is
316 /// known, `Unspecified` otherwise.
317 ///
318 /// # Errors
319 /// Returns an [`Error`] if authentication or the RPC fails.
320 pub async fn get_package_status(&self, package_id: &str) -> Result<lapi::PackageStatus> {
321 let package_id = package_id.to_string();
322 telemetry::instrument("get_package_status", TRANSPORT_GRPC, async move {
323 self.with_retry(|| {
324 let package_id = package_id.clone();
325 async move {
326 let mut client = PackageServiceClient::new(self.intercepted().await?);
327 let response = client
328 .get_package_status(lapi::GetPackageStatusRequest { package_id })
329 .await?
330 .into_inner();
331 Ok(response.package_status())
332 }
333 })
334 .await
335 })
336 .await
337 }
338
339 /// The specific parties the authenticated user may act as — the `CanActAs`
340 /// subset of [`Self::current_user_rights`]. Supersedes the ad-hoc party
341 /// discovery used in early tests.
342 ///
343 /// Note: a user granted the wildcard `CanActAsAnyParty` right may act as
344 /// *any* party yet carries no explicit `CanActAs` entries, so this returns
345 /// an empty list for such users — an empty result means "no specific
346 /// parties", not necessarily "cannot act". Inspect
347 /// [`Self::current_user_rights`] directly if you need the wildcard.
348 ///
349 /// # Errors
350 /// Returns an [`Error`] if the underlying rights RPC fails.
351 pub async fn acting_parties(&self) -> Result<Vec<String>> {
352 let rights = self.current_user_rights().await?;
353 Ok(rights
354 .into_iter()
355 .filter_map(|right| match right.kind {
356 Some(pb::right::Kind::CanActAs(can)) => Some(can.party),
357 _ => None,
358 })
359 .collect())
360 }
361}