From 37ce99298edc85e8a08fc7a8e819f2262a61c4fb Mon Sep 17 00:00:00 2001 From: Tim Bruijnzeels Date: Tue, 10 Aug 2021 15:45:54 +0200 Subject: [PATCH] Monitor child CA versions and send version to parent/repo (#622, #620) --- src/cli/client.rs | 9 +- src/cli/options.rs | 23 ++++- src/cli/report.rs | 7 +- src/commons/api/ca.rs | 128 +++++++++++++++++++++++---- src/commons/util/httpclient.rs | 13 ++- src/constants.rs | 1 + src/daemon/ca/manager.rs | 42 +++++++-- src/daemon/ca/status.rs | 153 +++++++++++++++++++++++---------- src/daemon/http/mod.rs | 17 ++++ src/daemon/http/server.rs | 68 +++++++++------ src/daemon/krillserver.rs | 41 +++++---- 11 files changed, 384 insertions(+), 118 deletions(-) diff --git a/src/cli/client.rs b/src/cli/client.rs index e66ef9be..9a7addeb 100644 --- a/src/cli/client.rs +++ b/src/cli/client.rs @@ -9,8 +9,8 @@ use rpki::uri; use crate::cli::options::KrillUserDetails; use crate::cli::report::{ApiResponse, ReportError}; use crate::commons::api::{ - AllCertAuthIssues, CaRepoDetails, CertAuthIssues, ChildCaInfo, ParentCaContact, ParentStatuses, PublisherDetails, - PublisherList, RepoStatus, Token, + AllCertAuthIssues, CaRepoDetails, CertAuthIssues, ChildCaInfo, ChildrenConnectionStats, ParentCaContact, + ParentStatuses, PublisherDetails, PublisherList, RepoStatus, Token, }; use crate::commons::bgp::BgpAnalysisAdvice; use crate::commons::remote::rfc8183; @@ -258,6 +258,11 @@ impl KrillClient { delete(&self.server, &self.token, &uri).await?; Ok(ApiResponse::Empty) } + CaCommand::ChildConnections(handle) => { + let uri = format!("api/v1/cas/{}/stats/children/connections", handle); + let stats: ChildrenConnectionStats = get_json(&self.server, &self.token, &uri).await?; + Ok(ApiResponse::ChildrenStats(stats)) + } CaCommand::KeyRollInit(handle) => { let uri = format!("api/v1/cas/{}/keys/roll_init", handle); diff --git a/src/cli/options.rs b/src/cli/options.rs index b79bb468..b7a71841 100644 --- a/src/cli/options.rs +++ b/src/cli/options.rs @@ -9,9 +9,9 @@ use std::{env, fmt}; use bytes::Bytes; use clap::{App, Arg, ArgMatches, SubCommand}; -use rpki::uri; use rpki::repository::crypto::KeyIdentifier; use rpki::repository::x509::Time; +use rpki::uri; use crate::commons::crypto::{IdCert, SignSupport}; use crate::commons::remote::rfc8183; @@ -469,6 +469,15 @@ impl Options { app.subcommand(sub) } + fn make_cas_children_connections_sc<'a, 'b>(app: App<'a, 'b>) -> App<'a, 'b> { + let mut sub = SubCommand::with_name("connections").about("Show connections stats for children of a CA"); + + sub = Self::add_general_args(sub); + sub = Self::add_my_ca_arg(sub); + + app.subcommand(sub) + } + fn make_cas_children_sc<'a, 'b>(app: App<'a, 'b>) -> App<'a, 'b> { let mut sub = SubCommand::with_name("children").about("Manage children for a CA"); @@ -477,6 +486,7 @@ impl Options { sub = Self::make_cas_children_info_sc(sub); sub = Self::make_cas_children_remove_sc(sub); sub = Self::make_cas_children_response_sc(sub); + sub = Self::make_cas_children_connections_sc(sub); app.subcommand(sub) } @@ -1409,6 +1419,14 @@ impl Options { Ok(Options::make(general_args, command)) } + fn parse_matches_cas_children_connections(matches: &ArgMatches) -> Result { + let general_args = GeneralArgs::from_matches(matches)?; + let my_ca = Self::parse_my_ca(matches)?; + + let command = Command::CertAuth(CaCommand::ChildConnections(my_ca)); + Ok(Options::make(general_args, command)) + } + fn parse_matches_cas_children(matches: &ArgMatches) -> Result { if let Some(m) = matches.subcommand_matches("add") { Self::parse_matches_cas_children_add(m) @@ -1420,6 +1438,8 @@ impl Options { Self::parse_matches_cas_children_update(m) } else if let Some(m) = matches.subcommand_matches("remove") { Self::parse_matches_cas_children_remove(m) + } else if let Some(m) = matches.subcommand_matches("connections") { + Self::parse_matches_cas_children_connections(m) } else { Err(Error::UnrecognizedSubCommand) } @@ -2087,6 +2107,7 @@ pub enum CaCommand { ChildAdd(Handle, AddChildRequest), ChildUpdate(Handle, ChildHandle, UpdateChildRequest), ChildDelete(Handle, ChildHandle), + ChildConnections(Handle), // Key Management KeyRollInit(Handle), diff --git a/src/cli/report.rs b/src/cli/report.rs index aab9698b..c737993b 100644 --- a/src/cli/report.rs +++ b/src/cli/report.rs @@ -5,8 +5,8 @@ use serde::Serialize; use crate::commons::api::{ AllCertAuthIssues, CaCommandDetails, CaRepoDetails, CertAuthInfo, CertAuthIssues, CertAuthList, ChildCaInfo, - CommandHistory, ParentCaContact, ParentStatuses, PublisherDetails, PublisherList, RepoStatus, RoaDefinitions, - RtaList, RtaPrepResponse, ServerInfo, + ChildrenConnectionStats, CommandHistory, ParentCaContact, ParentStatuses, PublisherDetails, PublisherList, + RepoStatus, RoaDefinitions, RtaList, RtaPrepResponse, ServerInfo, }; use crate::commons::bgp::{BgpAnalysisAdvice, BgpAnalysisReport, BgpAnalysisSuggestion}; use crate::commons::remote::api::ClientInfos; @@ -36,6 +36,7 @@ pub enum ApiResponse { ParentStatuses(ParentStatuses), ChildInfo(ChildCaInfo), + ChildrenStats(ChildrenConnectionStats), PublisherDetails(PublisherDetails), PublisherList(PublisherList), @@ -81,6 +82,7 @@ impl ApiResponse { ApiResponse::ParentCaContact(contact) => Ok(Some(contact.report(fmt)?)), ApiResponse::ParentStatuses(statuses) => Ok(Some(statuses.report(fmt)?)), ApiResponse::ChildInfo(info) => Ok(Some(info.report(fmt)?)), + ApiResponse::ChildrenStats(stats) => Ok(Some(stats.report(fmt)?)), ApiResponse::PublisherList(list) => Ok(Some(list.report(fmt)?)), ApiResponse::PublisherDetails(details) => Ok(Some(details.report(fmt)?)), ApiResponse::RepoStats(stats) => Ok(Some(stats.report(fmt)?)), @@ -177,6 +179,7 @@ impl Report for CaCommandDetails {} impl Report for PublisherList {} impl Report for RepoStats {} +impl Report for ChildrenConnectionStats {} impl Report for PublisherDetails {} diff --git a/src/commons/api/ca.rs b/src/commons/api/ca.rs index e76c1546..21c3dece 100644 --- a/src/commons/api/ca.rs +++ b/src/commons/api/ca.rs @@ -1355,7 +1355,7 @@ impl ParentStatus { self.last_exchange = Some(ParentExchange { timestamp: Time::now().timestamp(), uri, - result: ParentExchangeResult::Failure(error), + result: ExchangeResult::Failure(error), }); self.set_next_exchange_plus_seconds(next_seconds); } @@ -1386,7 +1386,7 @@ impl ParentStatus { self.last_exchange = Some(ParentExchange { timestamp: Time::now().timestamp(), uri, - result: ParentExchangeResult::Success, + result: ExchangeResult::Success, }); self.set_next_exchange_plus_seconds(next_run_seconds); } @@ -1443,7 +1443,7 @@ impl RepoStatus { self.last_exchange = Some(ParentExchange { timestamp: Time::now().timestamp(), uri, - result: ParentExchangeResult::Failure(error), + result: ExchangeResult::Failure(error), }); self.next_exchange_before = (Time::now() + Duration::minutes(5)).timestamp(); } @@ -1452,7 +1452,7 @@ impl RepoStatus { self.last_exchange = Some(ParentExchange { timestamp: Time::now().timestamp(), uri, - result: ParentExchangeResult::Success, + result: ExchangeResult::Success, }); self.published = published; self.next_exchange_before = Self::now_plus_hours(next_hours); @@ -1462,7 +1462,7 @@ impl RepoStatus { self.last_exchange = Some(ParentExchange { timestamp: Time::now().timestamp(), uri, - result: ParentExchangeResult::Success, + result: ExchangeResult::Success, }); self.next_exchange_before = Self::now_plus_hours(next_hours); } @@ -1491,7 +1491,7 @@ impl fmt::Display for RepoStatus { pub struct ParentExchange { timestamp: i64, uri: ServiceUri, - result: ParentExchangeResult, + result: ExchangeResult, } impl ParentExchange { @@ -1503,41 +1503,139 @@ impl ParentExchange { &self.uri } - pub fn result(&self) -> &ParentExchangeResult { + pub fn result(&self) -> &ExchangeResult { &self.result } pub fn was_success(&self) -> bool { match &self.result { - ParentExchangeResult::Success => true, - ParentExchangeResult::Failure(_) => false, + ExchangeResult::Success => true, + ExchangeResult::Failure(_) => false, } } pub fn into_failure_opt(self) -> Option { match self.result { - ParentExchangeResult::Success => None, - ParentExchangeResult::Failure(error) => Some(error), + ExchangeResult::Success => None, + ExchangeResult::Failure(error) => Some(error), } } } #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] #[allow(clippy::large_enum_variant)] -pub enum ParentExchangeResult { +pub enum ExchangeResult { Success, Failure(ErrorResponse), } -impl fmt::Display for ParentExchangeResult { +impl fmt::Display for ExchangeResult { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { match self { - ParentExchangeResult::Success => write!(f, "success"), - ParentExchangeResult::Failure(e) => write!(f, "failure: {}", e.msg()), + ExchangeResult::Success => write!(f, "success"), + ExchangeResult::Failure(e) => write!(f, "failure: {}", e.msg()), } } } +//------------ ChildConnectionStats ------------------------------------------ + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct ChildrenConnectionStats { + children: Vec, +} + +impl ChildrenConnectionStats { + pub fn new(children: Vec) -> Self { + ChildrenConnectionStats { children } + } +} + +impl fmt::Display for ChildrenConnectionStats { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + if !self.children.is_empty() { + writeln!(f, "handle, user_agent, last_exchange, result")?; + for child in &self.children { + match &child.last_exchange { + None => { + writeln!(f, "{},n/a,never,n/a", child.handle)?; + } + Some(exchange) => { + let agent = exchange.user_agent.as_deref().unwrap_or(""); + let time = Time::new(Utc.timestamp(exchange.timestamp, 0)); + + writeln!( + f, + "{},{},{},{}", + child.handle, + agent, + time.to_rfc3339(), + exchange.result + )?; + } + } + } + } + Ok(()) + } +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct ChildConnectionStats { + handle: ChildHandle, + last_exchange: Option, +} + +impl ChildConnectionStats { + pub fn new(handle: ChildHandle, last_exchange: Option) -> Self { + ChildConnectionStats { handle, last_exchange } + } +} + +//------------ ChildStatus --------------------------------------------------- + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct ChildStatus { + last_exchange: Option, +} + +impl ChildStatus { + pub fn set_success(&mut self, user_agent: Option) { + self.last_exchange = Some(ChildExchange { + timestamp: Time::now().timestamp(), + result: ExchangeResult::Success, + user_agent, + }); + } + + pub fn set_failure(&mut self, user_agent: Option, error_response: ErrorResponse) { + self.last_exchange = Some(ChildExchange { + timestamp: Time::now().timestamp(), + result: ExchangeResult::Failure(error_response), + user_agent, + }); + } +} + +impl Default for ChildStatus { + fn default() -> Self { + ChildStatus { last_exchange: None } + } +} + +impl From for Option { + fn from(status: ChildStatus) -> Self { + status.last_exchange + } +} + +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +pub struct ChildExchange { + timestamp: i64, + result: ExchangeResult, + user_agent: Option, +} + //------------ CertAuthInfo -------------------------------------------------- /// This type represents the details of a CertAuth that need diff --git a/src/commons/util/httpclient.rs b/src/commons/util/httpclient.rs index 252e7973..74e292a7 100644 --- a/src/commons/util/httpclient.rs +++ b/src/commons/util/httpclient.rs @@ -12,7 +12,7 @@ use serde::Serialize; use crate::commons::api::{ErrorResponse, Token}; use crate::commons::util::file; -use crate::constants::{HTTP_CLIENT_TIMEOUT_SECS, KRILL_CLI_API_ENV, KRILL_HTTPS_ROOT_CERTS_ENV}; +use crate::constants::{HTTP_CLIENT_TIMEOUT_SECS, KRILL_CLI_API_ENV, KRILL_HTTPS_ROOT_CERTS_ENV, KRILL_VERSION}; const JSON_CONTENT: &str = "application/json"; @@ -184,17 +184,22 @@ pub async fn post_empty(uri: &str, token: Option<&Token>) -> Result<(), Error> { } } -/// Posts binary data, and expects a binary response. +/// Posts binary data, and expects a binary response. Includes the full krill version +/// as the user agent. Intended for sending RFC 6492 (provisioning) and 8181 (publication) +/// to the trusted parent or publication server. /// /// Note: Bytes may be empty if the post was successful, but the response was /// empty. -pub async fn post_binary(uri: &str, data: &Bytes, content_type: &str) -> Result { +pub async fn post_binary_with_full_ua(uri: &str, data: &Bytes, content_type: &str) -> Result { let body = data.to_vec(); if env::var(KRILL_CLI_API_ENV).is_ok() { report_post_and_exit(uri, None, None, PostBody::Bytes(&body)); } - let headers = headers(Some(content_type), None)?; + let mut headers = HeaderMap::new(); + headers.insert(USER_AGENT, HeaderValue::from_str(&format!("krill/{}", KRILL_VERSION))?); + headers.insert(CONTENT_TYPE, HeaderValue::from_str(content_type)?); + let res = client(uri).await?.post(uri).headers(headers).body(body).send().await?; match res.status() { diff --git a/src/constants.rs b/src/constants.rs index d158d9f6..4c457c97 100644 --- a/src/constants.rs +++ b/src/constants.rs @@ -76,6 +76,7 @@ pub const ID_CERTIFICATE_VALIDITY_YEARS: i32 = 15; pub const BGP_RIS_REFRESH_MINUTES: i64 = 60; pub const HTTP_CLIENT_TIMEOUT_SECS: u64 = 120; +pub const HTTP_USER_AGENT_TRUNCATE: usize = 256; // Will truncate received user-agent values at this size. pub const OPENID_CONNECT_HTTP_CLIENT_TIMEOUT_SECS: u64 = 30; pub const NO_RESOURCE: NoResourceType = NoResourceType; diff --git a/src/daemon/ca/manager.rs b/src/daemon/ca/manager.rs index 5242698b..0288636e 100644 --- a/src/daemon/ca/manager.rs +++ b/src/daemon/ca/manager.rs @@ -15,7 +15,6 @@ use rpki::uri; use crate::{ commons::{ actor::Actor, - api::rrdp::PublishElement, api::{ self, AddChildRequest, Base64, CaCommandDetails, CaCommandResult, CertAuthList, CertAuthSummary, ChildCaInfo, ChildHandle, CommandHistory, CommandHistoryCriteria, Entitlements, Handle, IssuanceRequest, @@ -23,6 +22,7 @@ use crate::{ RcvdCert, RepoStatus, RepositoryContact, ResourceClassName, ResourceSet, RevocationRequest, RevocationResponse, RtaName, StoredEffect, UpdateChildRequest, }, + api::{rrdp::PublishElement, ChildrenConnectionStats}, crypto::{IdCert, KrillSigner, ProtocolCms, ProtocolCmsBuilder}, error::Error, eventsourcing::{Aggregate, AggregateStore, Command, CommandKey}, @@ -445,6 +445,11 @@ impl CaManager { ca.get_child(child).map(|details| details.clone().into()) } + /// Show the (connection) stats for children under a CA. + pub async fn ca_stats_child_connections(&self, ca: &Handle) -> KrillResult { + self.status_store.lock().await.get_stats_child_connections(ca).await + } + /// Show a contact for a child. pub async fn ca_parent_contact( &self, @@ -515,7 +520,13 @@ impl CaManager { /// - validates the request /// - processes the child request /// - signs a response and returns the bytes - pub async fn rfc6492(&self, ca_handle: &Handle, msg_bytes: Bytes, actor: &Actor) -> KrillResult { + pub async fn rfc6492( + &self, + ca_handle: &Handle, + msg_bytes: Bytes, + user_agent: Option, + actor: &Actor, + ) -> KrillResult { let ca = self.get_ca(ca_handle).await?; let msg = match ProtocolCms::decode(msg_bytes.as_ref(), false) { @@ -540,22 +551,23 @@ impl CaManager { let (res, should_log_cms) = match content { rfc6492::Content::Qry(rfc6492::Qry::Revoke(req)) => { let res = self.revoke(ca_handle, child.clone(), req, actor).await?; - let msg = rfc6492::Message::revoke_response(child, recipient, res); + let msg = rfc6492::Message::revoke_response(child.clone(), recipient, res); (self.wrap_rfc6492_response(ca_handle, msg).await, true) } rfc6492::Content::Qry(rfc6492::Qry::List) => { let entitlements = self.list(ca_handle, &child).await?; - let msg = rfc6492::Message::list_response(child, recipient, entitlements); + let msg = rfc6492::Message::list_response(child.clone(), recipient, entitlements); (self.wrap_rfc6492_response(ca_handle, msg).await, false) } rfc6492::Content::Qry(rfc6492::Qry::Issue(req)) => { let res = self.issue(ca_handle, &child, req, actor).await?; - let msg = rfc6492::Message::issue_response(child, recipient, res); + let msg = rfc6492::Message::issue_response(child.clone(), recipient, res); (self.wrap_rfc6492_response(ca_handle, msg).await, true) } _ => (Err(Error::custom("Unsupported RFC6492 message")), true), }; + // Log CMS messages if needed, and if enabled by config (this is a no-op if it isn't) match &res { Ok(reply_bytes) => { if should_log_cms { @@ -569,6 +581,24 @@ impl CaManager { } } + // Set child status + match &res { + Ok(_) => { + self.status_store + .lock() + .await + .set_child_success(ca.handle(), &child, user_agent) + .await?; + } + Err(e) => { + self.status_store + .lock() + .await + .set_child_failure(ca.handle(), &child, user_agent, e) + .await?; + } + } + res } @@ -1589,7 +1619,7 @@ impl CaManager { let uri = service_uri.to_string(); - let res = httpclient::post_binary(&uri, &signed_msg, content_type) + let res = httpclient::post_binary_with_full_ua(&uri, &signed_msg, content_type) .await .map_err(Error::HttpClientError)?; diff --git a/src/daemon/ca/status.rs b/src/daemon/ca/status.rs index 46ed85cc..2c7e475f 100644 --- a/src/daemon/ca/status.rs +++ b/src/daemon/ca/status.rs @@ -1,15 +1,18 @@ -use std::path::Path; +use std::{collections::HashMap, path::Path}; use tokio::sync::RwLock; -use crate::commons::api::{ - rrdp::PublishElement, Entitlements, ErrorResponse, Handle, ParentHandle, ParentStatuses, RepoStatus, +use crate::commons::{ + api::{ + rrdp::PublishElement, ChildConnectionStats, ChildHandle, ChildStatus, ChildrenConnectionStats, Entitlements, + ErrorResponse, Handle, ParentHandle, ParentStatuses, RepoStatus, + }, + error::Error, + eventsourcing::{KeyStoreKey, KeyValueStore}, + remote::rfc8183::ServiceUri, + util::httpclient, + KrillResult, }; -use crate::commons::error::Error; -use crate::commons::eventsourcing::{KeyStoreKey, KeyValueStore}; -use crate::commons::remote::rfc8183::ServiceUri; -use crate::commons::util::httpclient; -use crate::commons::KrillResult; //------------ CaStatus ------------------------------------------------------ @@ -17,6 +20,20 @@ use crate::commons::KrillResult; struct CaStatus { repo: RepoStatus, parents: ParentStatuses, + #[serde(skip_serializing_if = "HashMap::is_empty", default = "HashMap::new")] + children: HashMap, +} + +impl CaStatus { + pub fn get_children_connection_stats(&self) -> ChildrenConnectionStats { + let children = self + .children + .clone() + .into_iter() + .map(|(handle, status)| ChildConnectionStats::new(handle, status.into())) + .collect(); + ChildrenConnectionStats::new(children) + } } impl Default for CaStatus { @@ -24,6 +41,7 @@ impl Default for CaStatus { CaStatus { repo: RepoStatus::default(), parents: ParentStatuses::default(), + children: HashMap::new(), } } } @@ -51,12 +69,6 @@ impl StatusStore { Ok(self.store.get(&Self::status_key(ca))?.unwrap_or_default()) } - /// Save the status for a CA - fn set_ca_status(&self, ca: &Handle, status: &CaStatus) -> KrillResult<()> { - self.store.store(&Self::status_key(ca), status)?; - Ok(()) - } - pub async fn get_parent_statuses(&self, ca: &Handle) -> KrillResult { let _lock = self.lock.read().await; let status = self.get_ca_status(ca)?; @@ -77,15 +89,14 @@ impl StatusStore { error: &Error, next_run_seconds: i64, ) -> KrillResult<()> { - let _lock = self.lock.write().await; - let mut status = self.get_ca_status(ca)?; - let error_response = Self::error_to_error_res(&error); - status - .parents - .set_failure(parent, uri, error_response, next_run_seconds); - self.set_ca_status(ca, &status) + self.update_ca_status(ca, |status| { + status + .parents + .set_failure(parent, uri, error_response, next_run_seconds) + }) + .await } pub async fn set_parent_last_updated( @@ -95,10 +106,10 @@ impl StatusStore { uri: &ServiceUri, next_run_seconds: i64, ) -> KrillResult<()> { - let _lock = self.lock.write().await; - let mut status = self.get_ca_status(ca)?; - status.parents.set_last_updated(parent, uri, next_run_seconds); - self.set_ca_status(ca, &status) + self.update_ca_status(ca, |status| { + status.parents.set_last_updated(parent, uri, next_run_seconds) + }) + .await } pub async fn set_parent_entitlements( @@ -109,37 +120,57 @@ impl StatusStore { entitlements: &Entitlements, next_run_seconds: i64, ) -> KrillResult<()> { - let _lock = self.lock.write().await; - let mut status = self.get_ca_status(ca)?; - status - .parents - .set_entitlements(parent, uri, entitlements, next_run_seconds); - self.set_ca_status(ca, &status) + self.update_ca_status(ca, |status| { + status + .parents + .set_entitlements(parent, uri, entitlements, next_run_seconds) + }) + .await } pub async fn remove_parent(&self, ca: &Handle, parent: &ParentHandle) -> KrillResult<()> { - let _lock = self.lock.write().await; - let mut status = self.get_ca_status(ca)?; - status.parents.remove(parent); + self.update_ca_status(ca, |status| status.parents.remove(parent)).await + } - Ok(()) + pub async fn set_child_success( + &self, + ca: &Handle, + child: &ChildHandle, + user_agent: Option, + ) -> KrillResult<()> { + self.update_ca_child_status(ca, child, |status| status.set_success(user_agent)) + .await + } + + pub async fn set_child_failure( + &self, + ca: &Handle, + child: &ChildHandle, + user_agent: Option, + error: &Error, + ) -> KrillResult<()> { + let error_response = Self::error_to_error_res(&error); + + self.update_ca_child_status(ca, child, |status| status.set_failure(user_agent, error_response)) + .await + } + + pub async fn get_stats_child_connections(&self, ca: &Handle) -> KrillResult { + let _lock = self.lock.read().await; + let status = self.get_ca_status(ca)?; + + Ok(status.get_children_connection_stats()) } pub async fn set_status_repo_failure(&self, ca: &Handle, uri: ServiceUri, error: &Error) -> KrillResult<()> { - let _lock = self.lock.write().await; - let mut status = self.get_ca_status(ca)?; - let error_response = Self::error_to_error_res(&error); - - status.repo.set_failure(uri, error_response); - self.set_ca_status(ca, &status) + self.update_ca_status(ca, |status| status.repo.set_failure(uri, error_response)) + .await } pub async fn set_status_repo_success(&self, ca: &Handle, uri: ServiceUri, next_hours: i64) -> KrillResult<()> { - let _lock = self.lock.write().await; - let mut status = self.get_ca_status(ca)?; - status.repo.set_last_updated(uri, next_hours); - self.set_ca_status(ca, &status) + self.update_ca_status(ca, |status| status.repo.set_last_updated(uri, next_hours)) + .await } pub async fn set_status_repo_published( @@ -149,10 +180,38 @@ impl StatusStore { published: Vec, next_hours: i64, ) -> KrillResult<()> { + self.update_ca_status(ca, |status| status.repo.set_published(uri, published, next_hours)) + .await + } + + async fn update_ca_status(&self, ca: &Handle, op: F) -> KrillResult<()> + where + F: FnOnce(&mut CaStatus), + { let _lock = self.lock.write().await; let mut status = self.get_ca_status(ca)?; - status.repo.set_published(uri, published, next_hours); - self.set_ca_status(ca, &status) + op(&mut status); + + self.store.store(&Self::status_key(ca), &status)?; + + Ok(()) + } + + async fn update_ca_child_status(&self, ca: &Handle, child: &ChildHandle, op: F) -> KrillResult<()> + where + F: FnOnce(&mut ChildStatus), + { + self.update_ca_status(ca, |status| { + let mut child_status = match status.children.get_mut(child) { + Some(child_status) => child_status, + None => { + status.children.insert(child.clone(), ChildStatus::default()); + status.children.get_mut(child).unwrap() + } + }; + op(&mut child_status) + }) + .await } fn error_to_error_res(error: &Error) -> ErrorResponse { diff --git a/src/daemon/http/mod.rs b/src/daemon/http/mod.rs index 7e7a85d8..9dc0374c 100644 --- a/src/daemon/http/mod.rs +++ b/src/daemon/http/mod.rs @@ -1,3 +1,4 @@ +use hyper::header::USER_AGENT; use serde::de::DeserializeOwned; use std::io; use std::str::FromStr; @@ -16,6 +17,7 @@ use crate::commons::{ actor::{Actor, ActorDef}, KrillResult, }; +use crate::constants::HTTP_USER_AGENT_TRUNCATE; use crate::daemon::auth::LoggedInUser; use crate::daemon::http::server::State; @@ -353,6 +355,21 @@ impl Request { self.request.headers() } + pub fn user_agent(&self) -> Option { + match self.headers().get(&USER_AGENT) { + None => None, + Some(value) => value.to_str().ok().map(|s| { + // Note: HeaderValue.to_str() only returns ok in case the value is plain + // ascii so it's safe to treat bytes as characters here. + if s.len() > HTTP_USER_AGENT_TRUNCATE { + s[..HTTP_USER_AGENT_TRUNCATE].to_string() + } else { + s.to_string() + } + }), + } + } + pub async fn upgrade_from_anonymous(&mut self, actor_def: ActorDef) { if self.actor.is_anonymous() { self.actor = self.state.actor_from_def(actor_def); diff --git a/src/daemon/http/server.rs b/src/daemon/http/server.rs index 6986d049..00785c29 100644 --- a/src/daemon/http/server.rs +++ b/src/daemon/http/server.rs @@ -77,23 +77,19 @@ fn test_data_dir_or_die(config_item: &str, dir: &Path) { let test_file = dir.join("test"); if let Err(e) = file::save(b"test", &test_file) { - print_write_error_hint_and_die( - format!( - "Cannot write to dir '{}' for configuration setting '{}', Error: {}", - dir.to_string_lossy(), - config_item, - e - ) - ); + print_write_error_hint_and_die(format!( + "Cannot write to dir '{}' for configuration setting '{}', Error: {}", + dir.to_string_lossy(), + config_item, + e + )); } else if let Err(e) = file::delete_file(&test_file) { - print_write_error_hint_and_die( - format!( - "Cannot delete test file '{}' in dir for configuration setting '{}', Error: {}", - test_file.to_string_lossy(), - config_item, - e - ) - ); + print_write_error_hint_and_die(format!( + "Cannot delete test file '{}' in dir for configuration setting '{}', Error: {}", + test_file.to_string_lossy(), + config_item, + e + )); } } @@ -665,13 +661,14 @@ pub async fn rfc6492(req: Request) -> RoutingResult { let actor = req.actor(); let state = req.state().clone(); + let user_agent = req.user_agent(); let bytes = match req.rfc6492_bytes().await { Ok(bytes) => bytes, Err(e) => return render_error(e), }; let krill_server = state; - match krill_server.rfc6492(ca, bytes, &actor).await { + match krill_server.rfc6492(ca, bytes, user_agent, &actor).await { Ok(bytes) => Ok(HttpResponse::rfc6492(bytes.to_vec())), Err(e) => render_error(e), } @@ -871,7 +868,10 @@ async fn api_cas(req: Request, path: &mut RequestPath) -> RoutingResult { Some("parents") => api_ca_parents(req, path, ca).await, Some("repo") => api_ca_repo(req, path, ca).await, Some("routes") => api_ca_routes(req, path, ca).await, + Some("stats") => api_ca_stats(req, path, ca).await, + Some("rta") => api_ca_rta(req, path, ca).await, + _ => render_unknown_method(), } }), @@ -939,6 +939,16 @@ async fn api_ca_routes(req: Request, path: &mut RequestPath, ca: Handle) -> Rout } } +async fn api_ca_stats(req: Request, path: &mut RequestPath, ca: Handle) -> RoutingResult { + match path.next() { + Some("children") => match path.next() { + Some("connections") => api_ca_stats_child_connections(req, ca).await, + _ => render_unknown_method(), + }, + _ => render_unknown_method(), + } +} + async fn api_publication_server(req: Request, path: &mut RequestPath) -> RoutingResult { match path.next() { Some("publishers") => api_publishers(req, path).await, @@ -991,7 +1001,8 @@ pub async fn api_stale_publishers(req: Request, seconds: Option<&str>) -> Routin let seconds = seconds.unwrap_or(""); match i64::from_str(seconds) { Ok(seconds) => render_json_res( - req.state().repo_stats() + req.state() + .repo_stats() .map(|stats| PublisherList::build(&stats.stale_publishers(seconds))), ), Err(_) => render_error(Error::ApiInvalidSeconds), @@ -1003,7 +1014,9 @@ pub async fn api_stale_publishers(req: Request, seconds: Option<&str>) -> Routin pub async fn api_list_pbl(req: Request) -> RoutingResult { aa!(req, Permission::PUB_LIST, { render_json_res( - req.state().publishers().map(|publishers| PublisherList::build(&publishers)), + req.state() + .publishers() + .map(|publishers| PublisherList::build(&publishers)), ) }) } @@ -1101,6 +1114,15 @@ async fn api_ca_child_show(req: Request, ca: Handle, child: ChildHandle) -> Rout ) } +async fn api_ca_stats_child_connections(req: Request, ca: Handle) -> RoutingResult { + aa!( + req, + Permission::CA_READ, + ca.clone(), + render_json_res(req.state().ca_stats_child_connections(&ca).await) + ) +} + async fn api_ca_parent_contact(req: Request, ca: Handle, child: ChildHandle) -> RoutingResult { aa!( req, @@ -1431,9 +1453,7 @@ async fn api_ca_parent_add_or_update(req: Request, ca: Handle, parent_override: }; match extract_parent_ca_req(&ca, bytes, parent_override) { - Ok(parent_req) => render_empty_res( - server.ca_parent_add_or_update(ca, parent_req, &actor).await, - ), + Ok(parent_req) => render_empty_res(server.ca_parent_add_or_update(ca, parent_req, &actor).await), Err(e) => render_error(e), } }) @@ -1577,9 +1597,7 @@ async fn api_ca_routes_analysis(req: Request, path: &mut RequestPath, ca: Handle let server = req.state().clone(); match req.json().await { Err(e) => render_error(e), - Ok(resources) => { - render_json_res(server.ca_routes_bgp_suggest(&ca, Some(resources)).await) - } + Ok(resources) => render_json_res(server.ca_routes_bgp_suggest(&ca, Some(resources)).await), } } _ => render_unknown_method(), diff --git a/src/daemon/krillserver.rs b/src/daemon/krillserver.rs index e78620bc..92db5c44 100644 --- a/src/daemon/krillserver.rs +++ b/src/daemon/krillserver.rs @@ -7,14 +7,10 @@ use bytes::Bytes; use chrono::Duration; use rpki::{ - repository::{ - cert::Cert, - x509::Time, - }, + repository::{cert::Cert, x509::Time}, uri, }; -use crate::commons::actor::{Actor, ActorDef}; use crate::commons::api::{ AddChildRequest, AllCertAuthIssues, CaCommandDetails, CaRepoDetails, CertAuthInfo, CertAuthInit, CertAuthIssues, CertAuthList, CertAuthStats, ChildCaInfo, ChildHandle, CommandHistory, CommandHistoryCriteria, Handle, ListReply, @@ -26,6 +22,10 @@ use crate::commons::bgp::{BgpAnalyser, BgpAnalysisReport, BgpAnalysisSuggestion} use crate::commons::crypto::KrillSigner; use crate::commons::eventsourcing::CommandKey; use crate::commons::remote::rfc8183; +use crate::commons::{ + actor::{Actor, ActorDef}, + api::ChildrenConnectionStats, +}; use crate::commons::{KrillEmptyResult, KrillResult}; use crate::constants::*; #[cfg(feature = "multi-user")] @@ -433,11 +433,16 @@ impl KrillServer { Ok(()) } - /// Show details for a child under the TA. - pub async fn ca_child_show(&self, parent: &ParentHandle, child: &ChildHandle) -> KrillResult { - let child = self.ca_manager.ca_show_child(parent, child).await?; + /// Show details for a child under the CA. + pub async fn ca_child_show(&self, ca: &Handle, child: &ChildHandle) -> KrillResult { + let child = self.ca_manager.ca_show_child(ca, child).await?; Ok(child) } + + /// Show children stats under the CA. + pub async fn ca_stats_child_connections(&self, ca: &Handle) -> KrillResult { + self.ca_manager.ca_stats_child_connections(ca).await + } } /// # Being a child @@ -457,15 +462,13 @@ impl KrillServer { ) -> KrillEmptyResult { let parent = parent_req.handle(); let contact = parent_req.contact(); - self.ca_manager.get_entitlements_from_contact(&ca, parent, contact, false).await?; + self.ca_manager + .get_entitlements_from_contact(&ca, parent, contact, false) + .await?; - Ok(self - .ca_manager - .ca_parent_add_or_update(ca, parent_req, actor) - .await?) + Ok(self.ca_manager.ca_parent_add_or_update(ca, parent_req, actor).await?) } - pub async fn ca_parent_remove(&self, handle: Handle, parent: ParentHandle, actor: &Actor) -> KrillEmptyResult { Ok(self.ca_manager.ca_parent_remove(handle, parent, actor).await?) } @@ -646,8 +649,14 @@ impl KrillServer { .await?) } - pub async fn rfc6492(&self, handle: Handle, msg_bytes: Bytes, actor: &Actor) -> KrillResult { - Ok(self.ca_manager.rfc6492(&handle, msg_bytes, actor).await?) + pub async fn rfc6492( + &self, + handle: Handle, + msg_bytes: Bytes, + user_agent: Option, + actor: &Actor, + ) -> KrillResult { + Ok(self.ca_manager.rfc6492(&handle, msg_bytes, user_agent, actor).await?) } }