From 38464aa2cfe9049a155fdadd47a8a503580ffca7 Mon Sep 17 00:00:00 2001 From: Tim Bruijnzeels Date: Fri, 19 Jul 2019 21:20:18 +0200 Subject: [PATCH] Introducing message queue for asynchronous triggered processes: request certs, publish. (work in progress for #13) --- Cargo.toml | 2 +- ca/Cargo.toml | 28 --- ca/README.md | 7 - ca/src/lib.rs | 273 ----------------------- commons/src/api/admin.rs | 32 ++- daemon/Cargo.toml | 3 - daemon/defaults/krill.conf | 2 +- {ca/src => daemon/src/ca}/ca.rs | 31 +-- {ca/src => daemon/src/ca}/caserver.rs | 279 ++++++++++++++++++++++-- daemon/src/ca/mod.rs | 21 ++ {ca/src => daemon/src/ca}/publishing.rs | 62 +++--- {ca/src => daemon/src/ca}/signing.rs | 0 daemon/src/config.rs | 8 +- daemon/src/endpoints.rs | 45 +++- daemon/src/krillserver.rs | 83 ++++++- daemon/src/lib.rs | 5 +- daemon/src/mq.rs | 124 +++++++++++ daemon/tests/ca_under_ta.rs | 5 +- 18 files changed, 589 insertions(+), 421 deletions(-) delete mode 100644 ca/Cargo.toml delete mode 100644 ca/README.md delete mode 100644 ca/src/lib.rs rename {ca/src => daemon/src/ca}/ca.rs (98%) rename {ca/src => daemon/src/ca}/caserver.rs (67%) create mode 100644 daemon/src/ca/mod.rs rename {ca/src => daemon/src/ca}/publishing.rs (74%) rename {ca/src => daemon/src/ca}/signing.rs (100%) create mode 100644 daemon/src/mq.rs diff --git a/Cargo.toml b/Cargo.toml index 7ee4c5c5..42e7f4ff 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = [ "ca", "client", "commons", "daemon", "pubc", "pubd" ] +members = [ "client", "commons", "daemon", "pubc", "pubd" ] [patch.crates-io] rpki = { git="https://github.com/NLnetLabs/rpki-rs.git" } diff --git a/ca/Cargo.toml b/ca/Cargo.toml deleted file mode 100644 index f1c8d66d..00000000 --- a/ca/Cargo.toml +++ /dev/null @@ -1,28 +0,0 @@ -[package] -name = "krill_ca" -version = "0.2.0" -authors = ["Tim Bruijnzeels ", "Martin Hoffmann "] - -[dependencies] -bcder = "^0.3" -base64 = "^0.10" -bytes = "^0.4" -chrono = "^0.4" -derive_more = "^0.13" -hex = "^0.3" -rand = "^0.6" -rpki = "^0.5" -serde = { version = "^1.0", features = ["derive" ] } -serde_json = "^1.0" - -# Logging -fern = "^0.5" -log = "^0.4" - -[dependencies.krill_commons] -path = "../commons" -version = "0.2.0" - -[features] -default = [] -extra-debug = [ "rpki/extra-debug" ] \ No newline at end of file diff --git a/ca/README.md b/ca/README.md deleted file mode 100644 index 79ddc29b..00000000 --- a/ca/README.md +++ /dev/null @@ -1,7 +0,0 @@ -# Krill Certificate Authority - -This module provides support for RPKI CA functions. - -## License - -This software is distributed under the Mozilla Public License 2.0. See the LICENSE file included. diff --git a/ca/src/lib.rs b/ca/src/lib.rs deleted file mode 100644 index e6fcbc43..00000000 --- a/ca/src/lib.rs +++ /dev/null @@ -1,273 +0,0 @@ -extern crate base64; -extern crate bytes; -extern crate chrono; -extern crate core; -#[macro_use] extern crate derive_more; -extern crate hex; -#[macro_use] extern crate log; -extern crate rand; -#[macro_use] extern crate serde; -extern crate serde_json; - -extern crate bcder; -extern crate rpki; -extern crate krill_commons; - -mod ca; -pub use ca::ta_handle; -pub use ca::Error as CaError; - -mod caserver; -pub use self::caserver::CaServer; -pub use self::caserver::Error as CaServerError; - -mod signing; -pub use self::signing::CaSigner; -pub use self::signing::CaSignSupport; - -mod publishing; -pub use self::publishing::PubClients; -pub use self::publishing::Error as PubClientError; - - -//------------ Tests --------------------------------------------------------- - -#[cfg(test)] -mod tests { - use std::path::PathBuf; - use std::sync::{Arc, RwLock}; - - use krill_commons::api::{DFLT_CLASS, IssuanceRequest}; - use krill_commons::api::admin::{ - Handle, - Token, - ParentCaContact - }; - use krill_commons::api::ca::{ - RepoInfo, - ResourceSet, - RcvdCert - }; - use krill_commons::eventsourcing::{ - Aggregate, - AggregateStore, - DiskAggregateStore - }; - use krill_commons::util::softsigner::OpenSslSigner; - use krill_commons::util::test::{ - sub_dir, - https, - rsync, - test_under_tmp, - }; - - use crate::ca::{ - ta_handle, - CA_NS, - CertAuth, - CaIniDet, - CaCmdDet, - CaEvtDet, - }; - - fn signer(temp_dir: &PathBuf) -> OpenSslSigner { - let signer_dir = sub_dir(temp_dir); - OpenSslSigner::build(&signer_dir).unwrap() - } - - #[test] - fn init_ta() { - test_under_tmp(|d| { - let ca_store = DiskAggregateStore::>::new( - &d, CA_NS - ).unwrap(); - - let ta_repo_info = { - let base_uri = rsync("rsync://localhost/repo/ta/"); - let rrdp_uri = https("https://localhost/repo/notifcation.xml"); - RepoInfo::new(base_uri, rrdp_uri) - }; - - let ta_handle = ta_handle(); - - - let ta_uri = https("https://localhost/tal/ta.cer"); - let ta_aia = rsync("rsync://localhost/repo/ta.cer"); - - let signer = signer(&d); - let signer = Arc::new(RwLock::new(signer)); - - // - // --- Create TA and publish - // - - let ta_ini = CaIniDet::init_ta( - &ta_handle, - ta_repo_info, - ta_aia, - vec![ta_uri], - - signer.clone() - ).unwrap(); - - ca_store.add(ta_ini).unwrap(); - let ta = ca_store.get_latest(&ta_handle).unwrap(); - - // - // --- Create Child CA - // - // Expect: - // - Child CA initialised - // - let child_handle = Handle::from("child"); - let child_token = Token::from("child"); - let child_rs = ResourceSet::from_strs("", "10.0.0.0/16", "").unwrap(); - - let ca_repo_info = { - let base_uri = rsync("rsync://localhost/repo/ca/"); - let rrdp_uri = https("https://localhost/repo/notifcation.xml"); - RepoInfo::new(base_uri, rrdp_uri) - }; - - let ca_ini = CaIniDet::init( - &child_handle, - child_token.clone(), - ca_repo_info, - signer.clone() - ).unwrap(); - - ca_store.add(ca_ini).unwrap(); - let child = ca_store.get_latest(&child_handle).unwrap(); - - // - // --- Add Child to TA - // - // Expect: - // - Child added to TA - // - - let cmd = CaCmdDet::add_child( - &ta_handle, - child_handle.clone(), - child_token.clone(), - None, - child_rs - ); - - let events = ta.process_command(cmd).unwrap(); - let ta = ca_store.update(&ta_handle, ta, events).unwrap(); - - // - // --- Add TA as parent to child CA - // - // Expect: - // - Parent added - // - - let parent = ParentCaContact::for_embedded( - ta_handle.clone(), - child_token.clone() - ); - - let add_parent = CaCmdDet::add_parent( - &child_handle, - ta_handle.as_str(), - parent - ); - - let events = child.process_command(add_parent).unwrap(); - let child = ca_store.update(&child_handle, child, events).unwrap(); - - // - // --- Get resource entitlements for Child and let it process - // - // Expect: - // - No change in TA (just read-only entitlements) - // - Resource Class (DFLT) added to child with pending key - // - Certificate requested by child - // - - let entitlements = ta.list(&child_handle, &child_token).unwrap(); - - let upd_ent = CaCmdDet::upd_entitlements( - &child_handle, - &ta_handle, - entitlements, - signer.clone() - ); - - let events = child.process_command(upd_ent).unwrap(); - assert_eq!(2, events.len()); // rc and csr - let req_evt = events[1].clone().into_details(); - let child = ca_store.update(&child_handle, child, events).unwrap(); - - let req = match req_evt { - CaEvtDet::CertificateRequested(req) => req, - _ => panic!("Expected Csr") - }; - - let (parent_info, issuance_req) = req.unwrap(); - let (class_name, limit, csr) = issuance_req.unwrap(); - assert_eq!("all", &class_name); - assert!(limit.is_empty()); - if let ParentCaContact::Embedded(handle, token) = parent_info { - assert_eq!(ta_handle, handle); - assert_eq!(child_token, token); - } else { - panic!("Expected embedded contact") - } - - // - // --- Send certificate request from child to TA - // - // Expect: - // - Certificate issued - // - Publication - // - - let request = IssuanceRequest::new( - DFLT_CLASS.to_string(), limit, csr - ); - - let ta_cmd = CaCmdDet::certify_child( - &ta_handle, - child_handle.clone(), - request, - child_token.clone(), - signer.clone() - ); - - let ta_events = ta.process_command(ta_cmd).unwrap(); - let issued_evt = ta_events[0].clone().into_details(); - let _ta = ca_store.update(&ta_handle, ta, ta_events).unwrap(); - - let issued = match issued_evt { - CaEvtDet::CertificateIssued(issued) => issued, - _ => panic!("Expected issued certificate.") - }; - - let (handle, issuance_res) = issued.unwrap(); - - let (class_name, _, _, issued) = issuance_res.unwrap(); - - assert_eq!(child_handle, handle); - assert_eq!(DFLT_CLASS, class_name); - - // - // --- Return issued certificate to child CA - // - // Expect: - // - Pending key activated - // - Publication - - let rcvd_cert = RcvdCert::from(issued); - - let upd_rcvd = CaCmdDet::upd_received_cert( - &child_handle, &ta_handle, DFLT_CLASS, rcvd_cert, signer.clone() - ); - - let events = child.process_command(upd_rcvd).unwrap(); - let _child = ca_store.update(&child_handle, child, events).unwrap(); - }) - } -} \ No newline at end of file diff --git a/commons/src/api/admin.rs b/commons/src/api/admin.rs index ce07a2f9..d67b9341 100644 --- a/commons/src/api/admin.rs +++ b/commons/src/api/admin.rs @@ -258,24 +258,31 @@ impl Eq for PublisherDetails {} #[derive(Clone, Debug, Deserialize, Serialize)] pub struct PublisherClientRequest { handle: Handle, - server_info: PubServerInfo + server_info: PubServerContact } impl PublisherClientRequest { - pub fn new(handle: Handle, server_info: PubServerInfo) -> Self { + pub fn new(handle: Handle, server_info: PubServerContact) -> Self { PublisherClientRequest { handle, server_info } } - pub fn for_krill( + pub fn embedded( + handle: Handle + ) -> Self { + let server_info = PubServerContact::embedded(); + PublisherClientRequest { handle, server_info } + } + + pub fn krill( handle: Handle, service_uri: uri::Https, token: Token ) -> Self { - let server_info = PubServerInfo::for_krill(service_uri, token); + let server_info = PubServerContact::for_krill(service_uri, token); PublisherClientRequest { handle, server_info } } - pub fn unwrap(self) -> (Handle, PubServerInfo) { + pub fn unwrap(self) -> (Handle, PubServerContact) { (self.handle, self.server_info) } } @@ -284,14 +291,21 @@ impl PublisherClientRequest { //------------ PubServerInfo ------------------------------------------------- #[derive(Clone, Debug, Deserialize, Display, Serialize)] -pub enum PubServerInfo { - #[display(fmt = "Krill at: {}", _0)] +pub enum PubServerContact { + #[display(fmt = "Embedded server.")] + Embedded, + + #[display(fmt = "Remote Krill at: {}, using token: {}", _0, _1)] KrillServer(uri::Https, Token) } -impl PubServerInfo { +impl PubServerContact { + pub fn embedded() -> Self { + PubServerContact::Embedded + } + pub fn for_krill(service_uri: uri::Https, token: Token) -> Self { - PubServerInfo::KrillServer(service_uri, token) + PubServerContact::KrillServer(service_uri, token) } } diff --git a/daemon/Cargo.toml b/daemon/Cargo.toml index 6604a353..52daa88e 100644 --- a/daemon/Cargo.toml +++ b/daemon/Cargo.toml @@ -37,9 +37,6 @@ xml-rs = "0.8.0" [build-dependencies] ignore = "^0.4" -[dependencies.krill_ca] -path = "../ca" -version = "0.2.0" [dependencies.krill_client] path = "../client" diff --git a/daemon/defaults/krill.conf b/daemon/defaults/krill.conf index a1dc9550..dbc1b87a 100644 --- a/daemon/defaults/krill.conf +++ b/daemon/defaults/krill.conf @@ -48,7 +48,7 @@ # for a file. If "file" is given, the "log_file" field needs to be given, too. # # Defaults to "syslog". -#log_type = "syslog" +log_type = "file" # Syslog facility # diff --git a/ca/src/ca.rs b/daemon/src/ca/ca.rs similarity index 98% rename from ca/src/ca.rs rename to daemon/src/ca/ca.rs index 809fbb59..c110283b 100644 --- a/ca/src/ca.rs +++ b/daemon/src/ca/ca.rs @@ -5,7 +5,6 @@ use std::ops::{Deref, DerefMut}; use std::sync::{Arc, RwLock}; use chrono::Duration; -use rand::Rng; use rpki::cert::{Cert, TbsCert, KeyUsage, Overclaim}; use rpki::crypto::{PublicKey, PublicKeyFormat}; @@ -27,18 +26,13 @@ use krill_commons::eventsourcing::{ StoredEvent, }; use krill_commons::util::softsigner::SignerKeyId; - -use crate::signing::{CaSigner, CaSignSupport}; use krill_commons::remote::id::IdCert; use krill_commons::remote::builder::IdCertBuilder; use krill_commons::remote::rfc8183::ChildRequest; -pub const CA_NS: &str = "cas"; -const TA_NAME: &str = "ta"; // reserved for TA +use crate::ca::signing::{CaSigner, CaSignSupport}; + -pub fn ta_handle() -> Handle { - Handle::from(TA_NAME) -} #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] struct Rfc8183Id { @@ -125,7 +119,7 @@ impl CaIniDet { key: &S::KeyId, signer: &S ) -> CaRes { - let serial: Serial = rand::thread_rng().gen::().into(); + let serial: Serial = Serial::random(signer).map_err(Error::signer)?; let pub_key = signer.get_key_info(&key).map_err(Error::signer)?; let name = pub_key.to_subject_name(); @@ -393,7 +387,7 @@ impl CaEvtDet { pub type CaCmd = SentCommand>; -type ParentHandle = Handle; +pub type ParentHandle = Handle; type ResourceClassName = String; #[derive(Clone, Debug)] @@ -647,7 +641,7 @@ impl Aggregate for CertAuth { let base_repo = details.2; let ca_type = details.3; - if ca_type == CaType::Child && handle == Handle::from(TA_NAME) { + if ca_type == CaType::Child && handle == Handle::from("ta") { return Err(Error::NameReservedTa) } @@ -687,10 +681,10 @@ impl Aggregate for CertAuth { self.children.insert(handle, details); }, CaEvtDet::CertificateIssued(cert_issued) => { - let (child_hndl, response) = cert_issued.unwrap(); + let (child_handle, response) = cert_issued.unwrap(); let (class_name, _, _, issued) = response.unwrap(); - let child = self.children.get_mut(&child_hndl).unwrap(); + let child = self.children.get_mut(&child_handle).unwrap(); child.add_cert(&class_name, issued); }, @@ -705,14 +699,9 @@ impl Aggregate for CertAuth { self.parents.get_mut(&parent).unwrap() .resources.insert(name, rc); } - CaEvtDet::CertificateRequested(req) => { - info!( - "Certificate requested for class {} from {}", - req.class_name(), - req.parent - ); - // do nothing, this should be picked up by listener and sent - // to parent + CaEvtDet::CertificateRequested(_req) => { + // do nothing, this is already sent to the parent, + // otherwise this event is not even saved. }, CaEvtDet::PendingKeyActivated(parent, class_name, cert) => { let parent = self.parent_mut(parent).unwrap(); diff --git a/ca/src/caserver.rs b/daemon/src/ca/caserver.rs similarity index 67% rename from ca/src/caserver.rs rename to daemon/src/ca/caserver.rs index 70d26856..191ed48d 100644 --- a/ca/src/caserver.rs +++ b/daemon/src/ca/caserver.rs @@ -8,33 +8,32 @@ use rpki::uri; use krill_commons::api::{DFLT_CLASS, Entitlements, IssuanceRequest, IssuanceResponse}; use krill_commons::api::admin::{AddChildRequest, Handle, ParentCaContact, AddParentRequest, Token, ChildAuthRequest}; use krill_commons::api::ca::{IssuedCert, RcvdCert, RepoInfo, CertAuthList, CertAuthSummary}; -use krill_commons::eventsourcing::{ - Aggregate, - AggregateStore, - AggregateStoreError, - DiskAggregateStore, -}; +use krill_commons::eventsourcing::{Aggregate, AggregateStore, AggregateStoreError, DiskAggregateStore}; use krill_commons::util::httpclient; use krill_commons::remote::builder::SignedMessageBuilder; +use krill_commons::remote::{rfc8183, rfc6492}; -use crate::ca::{ - self, - CA_NS, +use crate::ca::ca::{ CertAuth, CaCmdDet, CaIniDet, CaEvtDet, }; -use crate::{ - ta_handle, - CaSigner, - PubClients -}; -use krill_commons::remote::{rfc8183, rfc6492}; +use crate::ca::CaSigner; +use ca::{CaError, PubClients}; +use mq::EventQueueListener; +pub const CA_NS: &str = "cas"; +const TA_NAME: &str = "ta"; // reserved for TA + +pub fn ta_handle() -> Handle { + Handle::from(TA_NAME) +} + //------------ CaServer ------------------------------------------------------ +#[derive(Clone)] pub struct CaServer { signer: Arc>, ca_store: Arc>> @@ -47,10 +46,12 @@ impl CaServer { /// initialised. pub fn build( work_dir: &PathBuf, + events_queue: Arc, pub_clients: Arc, signer: S ) -> CaResult { let mut ca_store = DiskAggregateStore::>::new(work_dir, CA_NS)?; + ca_store.add_listener(events_queue); ca_store.add_listener(pub_clients); Ok(CaServer { @@ -316,7 +317,8 @@ impl CaServer { Ok(()) } - fn update_entitlements(&self, handle: &Handle) -> CaResult<(), S> { + /// Update entitlements for a CA + pub fn update_entitlements(&self, handle: &Handle) -> CaResult<(), S> { // Note: we can bail out on serious server side errors, indicating // a bug or data corruption issue on our side. However, we should @@ -480,7 +482,7 @@ pub enum Error { TrustAnchorNotInitialisedError, #[display(fmt = "{}", _0)] - CaError(ca::Error), + CaError(CaError), #[display(fmt = "CA {} was already initialised", _0)] DuplicateCa(String), @@ -511,8 +513,8 @@ impl From for Error { fn from(e: io::Error) -> Self { Error::IoError(e) } } -impl From for Error { - fn from(e: ca::Error) -> Self { Error::CaError(e) } +impl From for Error { + fn from(e: CaError) -> Self { Error::CaError(e) } } impl From for Error { @@ -526,8 +528,41 @@ impl From for Error { mod tests { use super::*; - use krill_commons::util::test; + + use std::path::PathBuf; + use std::sync::{Arc, RwLock}; + + use krill_commons::api::{DFLT_CLASS, IssuanceRequest}; + use krill_commons::api::admin::{ + Handle, + Token, + ParentCaContact + }; + use krill_commons::api::ca::{ + RepoInfo, + ResourceSet, + RcvdCert + }; + use krill_commons::eventsourcing::{ + Aggregate, + AggregateStore, + DiskAggregateStore + }; use krill_commons::util::softsigner::OpenSslSigner; + use krill_commons::util::test; + use krill_commons::util::test::{ + sub_dir, + https, + rsync, + test_under_tmp, + }; + + use crate::ca::PubClients; + + fn signer(temp_dir: &PathBuf) -> OpenSslSigner { + let signer_dir = sub_dir(temp_dir); + OpenSslSigner::build(&signer_dir).unwrap() + } #[test] fn add_ta() { @@ -535,8 +570,13 @@ mod tests { let signer = OpenSslSigner::build(&d).unwrap(); let pub_clients = Arc::new(PubClients::build(&d).unwrap()); + let event_queue = Arc::new(EventQueueListener::in_mem()); - let mut server = CaServer::::build(&d, pub_clients, signer).unwrap(); + let mut server = CaServer::::build( + &d, + event_queue, + pub_clients, + signer).unwrap(); let repo_info = { let base_uri = test::rsync("rsync://localhost/repo/ta/"); @@ -555,4 +595,201 @@ mod tests { }) } + + + #[test] + fn init_ta() { + test_under_tmp(|d| { + let ca_store = DiskAggregateStore::>::new( + &d, CA_NS + ).unwrap(); + + let ta_repo_info = { + let base_uri = rsync("rsync://localhost/repo/ta/"); + let rrdp_uri = https("https://localhost/repo/notifcation.xml"); + RepoInfo::new(base_uri, rrdp_uri) + }; + + let ta_handle = ta_handle(); + + + let ta_uri = https("https://localhost/tal/ta.cer"); + let ta_aia = rsync("rsync://localhost/repo/ta.cer"); + + let signer = signer(&d); + let signer = Arc::new(RwLock::new(signer)); + + // + // --- Create TA and publish + // + + let ta_ini = CaIniDet::init_ta( + &ta_handle, + ta_repo_info, + ta_aia, + vec![ta_uri], + + signer.clone() + ).unwrap(); + + ca_store.add(ta_ini).unwrap(); + let ta = ca_store.get_latest(&ta_handle).unwrap(); + + // + // --- Create Child CA + // + // Expect: + // - Child CA initialised + // + let child_handle = Handle::from("child"); + let child_token = Token::from("child"); + let child_rs = ResourceSet::from_strs("", "10.0.0.0/16", "").unwrap(); + + let ca_repo_info = { + let base_uri = rsync("rsync://localhost/repo/ca/"); + let rrdp_uri = https("https://localhost/repo/notifcation.xml"); + RepoInfo::new(base_uri, rrdp_uri) + }; + + let ca_ini = CaIniDet::init( + &child_handle, + child_token.clone(), + ca_repo_info, + signer.clone() + ).unwrap(); + + ca_store.add(ca_ini).unwrap(); + let child = ca_store.get_latest(&child_handle).unwrap(); + + // + // --- Add Child to TA + // + // Expect: + // - Child added to TA + // + + let cmd = CaCmdDet::add_child( + &ta_handle, + child_handle.clone(), + child_token.clone(), + None, + child_rs + ); + + let events = ta.process_command(cmd).unwrap(); + let ta = ca_store.update(&ta_handle, ta, events).unwrap(); + + // + // --- Add TA as parent to child CA + // + // Expect: + // - Parent added + // + + let parent = ParentCaContact::for_embedded( + ta_handle.clone(), + child_token.clone() + ); + + let add_parent = CaCmdDet::add_parent( + &child_handle, + ta_handle.as_str(), + parent + ); + + let events = child.process_command(add_parent).unwrap(); + let child = ca_store.update(&child_handle, child, events).unwrap(); + + // + // --- Get resource entitlements for Child and let it process + // + // Expect: + // - No change in TA (just read-only entitlements) + // - Resource Class (DFLT) added to child with pending key + // - Certificate requested by child + // + + let entitlements = ta.list(&child_handle, &child_token).unwrap(); + + let upd_ent = CaCmdDet::upd_entitlements( + &child_handle, + &ta_handle, + entitlements, + signer.clone() + ); + + let events = child.process_command(upd_ent).unwrap(); + assert_eq!(2, events.len()); // rc and csr + let req_evt = events[1].clone().into_details(); + let child = ca_store.update(&child_handle, child, events).unwrap(); + + let req = match req_evt { + CaEvtDet::CertificateRequested(req) => req, + _ => panic!("Expected Csr") + }; + + let (parent_info, issuance_req) = req.unwrap(); + let (class_name, limit, csr) = issuance_req.unwrap(); + assert_eq!("all", &class_name); + assert!(limit.is_empty()); + if let ParentCaContact::Embedded(handle, token) = parent_info { + assert_eq!(ta_handle, handle); + assert_eq!(child_token, token); + } else { + panic!("Expected embedded contact") + } + + // + // --- Send certificate request from child to TA + // + // Expect: + // - Certificate issued + // - Publication + // + + let request = IssuanceRequest::new( + DFLT_CLASS.to_string(), limit, csr + ); + + let ta_cmd = CaCmdDet::certify_child( + &ta_handle, + child_handle.clone(), + request, + child_token.clone(), + signer.clone() + ); + + let ta_events = ta.process_command(ta_cmd).unwrap(); + let issued_evt = ta_events[0].clone().into_details(); + let _ta = ca_store.update(&ta_handle, ta, ta_events).unwrap(); + + let issued = match issued_evt { + CaEvtDet::CertificateIssued(issued) => issued, + _ => panic!("Expected issued certificate.") + }; + + let (handle, issuance_res) = issued.unwrap(); + + let (class_name, _, _, issued) = issuance_res.unwrap(); + + assert_eq!(child_handle, handle); + assert_eq!(DFLT_CLASS, class_name); + + // + // --- Return issued certificate to child CA + // + // Expect: + // - Pending key activated + // - Publication + + let rcvd_cert = RcvdCert::from(issued); + + let upd_rcvd = CaCmdDet::upd_received_cert( + &child_handle, &ta_handle, DFLT_CLASS, rcvd_cert, signer.clone() + ); + + let events = child.process_command(upd_rcvd).unwrap(); + let _child = ca_store.update(&child_handle, child, events).unwrap(); + }) + } } \ No newline at end of file diff --git a/daemon/src/ca/mod.rs b/daemon/src/ca/mod.rs new file mode 100644 index 00000000..f562d748 --- /dev/null +++ b/daemon/src/ca/mod.rs @@ -0,0 +1,21 @@ +//! Certificate Authority related code. +//! + +mod ca; +pub use self::ca::CaEvt as CaEvt; +pub use self::ca::CaEvtDet as CaEvtDet; +pub use self::ca::CertAuth as CertAuth; +pub use self::ca::ParentHandle as ParentHandle; +pub use self::ca::Error as CaError; + +pub mod caserver; + +mod signing; +pub use self::signing::CaSigner; +pub use self::signing::CaSignSupport; + +mod publishing; +pub use self::publishing::PubClients; +pub use self::publishing::Error as PubClientError; + + diff --git a/ca/src/publishing.rs b/daemon/src/ca/publishing.rs similarity index 74% rename from ca/src/publishing.rs rename to daemon/src/ca/publishing.rs index 0285782e..ca22c12c 100644 --- a/ca/src/publishing.rs +++ b/daemon/src/ca/publishing.rs @@ -1,12 +1,11 @@ //! Supports publishing signed objects. -use std::{io, thread}; +use std::io; use std::path::PathBuf; -use krill_commons::api::ErrorCode; use krill_commons::api::admin::{ Handle, - PubServerInfo, + PubServerContact, PublisherClientRequest }; use krill_commons::api::ca::AllCurrentObjects; @@ -22,10 +21,9 @@ use krill_commons::eventsourcing::{ SentCommand, StoredEvent, }; -use krill_commons::util::httpclient; use crate::ca::{CertAuth, CaEvt, CaEvtDet}; -use crate::signing::CaSigner; +use crate::ca::signing::CaSigner; @@ -34,12 +32,12 @@ use crate::signing::CaSigner; pub type PubClientInit = StoredEvent; #[derive(Clone, Debug, Deserialize, Serialize)] -pub struct PubClientInitDetails(PubServerInfo); +pub struct PubClientInitDetails(PubServerContact); impl PubClientInitDetails { pub fn init( handle: &Handle, - server_info: PubServerInfo + server_info: PubServerContact ) -> PubClientInit { PubClientInit::new( handle, @@ -77,7 +75,7 @@ impl CommandDetails for PubClientCommandDetails { pub struct PubClient { handle: Handle, version: u64, - server: PubServerInfo + server: PubServerContact } impl Aggregate for PubClient { @@ -111,7 +109,7 @@ impl Aggregate for PubClient { } impl PubClient { - pub fn server_info(&self) -> &PubServerInfo { + pub fn server_info(&self) -> &PubServerContact { &self.server } } @@ -133,32 +131,36 @@ impl PubClients { &self, handle: &Handle, _current_objects: AllCurrentObjects, - delta: PublishDelta + _delta: PublishDelta ) { let client = self.store.get_latest(handle).unwrap(); match client.server_info() { - PubServerInfo::KrillServer(service_uri, token) => { - let uri = format!("{}publication/{}", service_uri, handle); - let token = token.clone(); - let handle = handle.clone(); - - thread::spawn(move ||{ - match httpclient::post_json(&uri, delta, Some(&token)) { - Err(httpclient::Error::ErrorWithJson(_code, err)) => { - let err: ErrorCode = err.into(); - if err == ErrorCode::ObjectAlreadyPresent || - err == ErrorCode::NoObjectForHashAndOrUri { - unimplemented!("https://github.com/NLnetLabs/krill/issues/42") - } else { - error!("{}", err) - } - }, - Err(e) => error!("{}", e), - Ok(()) => debug!("PubClients: published for {}", handle) - } - }); + PubServerContact::KrillServer(_uri, _token) => { + error!("Remote publication not implemented") +// let uri = format!("{}publication/{}", service_uri, handle); +// let token = token.clone(); +// let handle = handle.clone(); +// +// thread::spawn(move ||{ +// match httpclient::post_json(&uri, delta, Some(&token)) { +// Err(httpclient::Error::ErrorWithJson(_code, err)) => { +// let err: ErrorCode = err.into(); +// if err == ErrorCode::ObjectAlreadyPresent || +// err == ErrorCode::NoObjectForHashAndOrUri { +// unimplemented!("https://github.com/NLnetLabs/krill/issues/42") +// } else { +// error!("{}", err) +// } +// }, +// Err(e) => error!("{}", e), +// Ok(()) => info!("PubClients: published for {}", handle) +// } +// }); + }, + PubServerContact::Embedded => { + error!("Embedded publication not implemented") } } } diff --git a/ca/src/signing.rs b/daemon/src/ca/signing.rs similarity index 100% rename from ca/src/signing.rs rename to daemon/src/ca/signing.rs diff --git a/daemon/src/config.rs b/daemon/src/config.rs index 73d8f293..6e12f8da 100644 --- a/daemon/src/config.rs +++ b/daemon/src/config.rs @@ -160,14 +160,14 @@ impl Config { let data_dir = data_dir.clone(); let rsync_base = ConfigDefaults::rsync_base(); let rrdp_base_uri = ConfigDefaults::rrdp_base_uri(); - let log_level = LevelFilter::Debug; + let log_level = LevelFilter::Info; let log_type = LogType::File; let mut log_file = data_dir.clone(); log_file.push("krill.log"); let syslog_facility = ConfigDefaults::syslog_facility(); let auth_token = Token::from("secret"); - Config { + let c = Config { ip, port, use_ssl, @@ -179,7 +179,9 @@ impl Config { log_file, syslog_facility, auth_token - } + }; + c.init_logging().unwrap(); + c } /// Creates the config (at startup). Panics in case of issues. diff --git a/daemon/src/endpoints.rs b/daemon/src/endpoints.rs index eb669086..34b85566 100644 --- a/daemon/src/endpoints.rs +++ b/daemon/src/endpoints.rs @@ -12,17 +12,19 @@ use actix_web::web::{ use bytes::Bytes; use serde::Serialize; -use krill_ca::{CaError, CaServerError}; use krill_commons::api::{admin, publication, ErrorCode, ErrorResponse, IssuanceRequest}; use krill_commons::api::admin::{Handle, CertAuthInit, AddChildRequest, AddParentRequest}; use krill_commons::api::rrdp::VerificationError; use krill_commons::util::softsigner::OpenSslSigner; use krill_commons::remote::api::ClientInfo; use krill_commons::remote::sigmsg::SignedMessage; +use krill_commons::remote::rfc6492; use krill_pubd::publishers::PublisherError; use krill_pubd::repo::RrdpServerError; use crate::auth::Auth; +use crate::ca::CaError; +use crate::ca::caserver; use crate::http::server::AppServer; use crate::krillserver; @@ -203,10 +205,11 @@ pub fn handle_delta( handle: Path ) -> HttpResponse { let handle = handle.into_inner(); + let delta = delta.into_inner(); + debug!("Received delta request for {}", &handle); if_publication_allowed(&server, &handle, &auth, || { - render_empty_res(server.read().handle_delta(delta.into_inner(), &handle)) - } - ) + render_empty_res(server.read().handle_delta(delta, &handle)) + }) } /// Processes a list request sent to the API. @@ -217,7 +220,7 @@ pub fn handle_list( handle: Path ) -> HttpResponse { let handle = handle.into_inner(); - info!("Received list request for {}", &handle); + debug!("Received list request for {}", &handle); if_publication_allowed(&server, &handle, &auth, ||{ match server.read().handle_list(&handle) { Ok(list) => render_json(list), @@ -446,9 +449,29 @@ pub fn issue( /// pub fn rfc6492( server: web::Data, - auth: Auth, + parent: Path, + child: Path, + msg_bytes: Bytes, ) -> HttpResponse { - unimplemented!() + match SignedMessage::decode(msg_bytes, true) { + Ok(msg) => { + match server.read().rfc6492( + parent.into_inner(), + child.into_inner(), + msg + ) { + Ok(captured) => { + HttpResponse::build(StatusCode::OK) + .content_type(rfc6492::CONTENT_TYPE) + .body(captured.into_bytes()) + } + Err(e) => { + server_error(&Error::ServerError(e)) + } + } + } + Err(_) => server_error(&Error::CmsError) + } } @@ -554,10 +577,10 @@ impl ErrorToStatus for RrdpServerError { } } -impl ErrorToStatus for CaServerError { +impl ErrorToStatus for caserver::Error { fn status(&self) -> StatusCode { match self { - CaServerError::CaError(e) => e.status(), + caserver::Error::CaError(e) => e.status(), _ => StatusCode::INTERNAL_SERVER_ERROR } } @@ -643,10 +666,10 @@ impl ToErrorCode for RrdpServerError { } } -impl ToErrorCode for CaServerError { +impl ToErrorCode for caserver::Error { fn code(&self) -> ErrorCode { match self { - CaServerError::CaError(e) => e.code(), + caserver::Error::CaError(e) => e.code(), _ => ErrorCode::CaServerError } } diff --git a/daemon/src/krillserver.rs b/daemon/src/krillserver.rs index 80b921e3..2fa1311c 100644 --- a/daemon/src/krillserver.rs +++ b/daemon/src/krillserver.rs @@ -6,10 +6,9 @@ use std::sync::Arc; use bcder::Captured; use rpki::uri; -use krill_ca::{ta_handle, CaServer, CaServerError, PubClients, PubClientError}; use krill_commons::api::{publication, Entitlements, IssuanceRequest, IssuanceResponse}; use krill_commons::api::admin; -use krill_commons::api::admin::{Handle, Token, PubServerInfo, CertAuthInit, CertAuthPubMode, ParentCaContact, AddChildRequest, AddParentRequest}; +use krill_commons::api::admin::{Handle, Token, PubServerContact, CertAuthInit, CertAuthPubMode, ParentCaContact, AddChildRequest, AddParentRequest}; use krill_commons::api::ca::{TrustAnchorInfo, RcvdCert, CertAuthList, CertAuthInfo}; use krill_commons::api::publication::PublishRequest; use krill_commons::util::softsigner::{OpenSslSigner, SignerError}; @@ -22,8 +21,21 @@ use krill_commons::remote::sigmsg::SignedMessage; use krill_pubd::PubServer; use krill_pubd::publishers::Publisher; +use crate::ca::{ + PubClients, + PubClientError +}; +use crate::ca::caserver::{ + self, + ta_handle, + CaServer, +}; use crate::auth::{Auth, Authorizer}; use crate::republisher::Republisher; +use crate::mq::EventQueueListener; +use clokwerk::{Scheduler, ScheduleHandle, TimeUnits}; +use std::time::Duration; +use mq::QueueEvent; //------------ KrillServer --------------------------------------------------- @@ -65,7 +77,11 @@ pub struct KrillServer { // Responsible for republishing periodically #[allow(dead_code)] // keep this in scope - republisher: Republisher + republisher: Republisher, + + // Responsible for background tasks, e.g. re-publishing + #[allow(dead_code)] // just need to keep this in scope + tasks_thread: ScheduleHandle } @@ -99,13 +115,26 @@ impl KrillServer { let signer = OpenSslSigner::build(work_dir)?; let pub_clients = Arc::new(PubClients::build(work_dir)?); - let caserver = CaServer::build(work_dir, pub_clients.clone(), signer)?; + + let event_queue = Arc::new(EventQueueListener::in_mem()); + + let caserver = CaServer::build( + work_dir, + event_queue.clone(), + pub_clients.clone(), + signer + )?; let republisher = { let publish_uri = format!("{}api/v1/republish", service_uri); Republisher::new(publish_uri, token) }; + let tasks_thread = Self::make_event_lister( + event_queue, + caserver.clone() + ); + Ok( KrillServer { service_uri, @@ -115,11 +144,36 @@ impl KrillServer { pub_clients, caserver, proxy_server, - republisher + republisher, + tasks_thread } ) } + fn make_event_lister( + event_queue: Arc, + _caserver: CaServer + ) -> ScheduleHandle { + let mut scheduler = Scheduler::new(); + scheduler.every(5.seconds()).run(move || { + while let Some(evt) = event_queue.pop() { + match evt { + QueueEvent::Delta(handle, _delta) => { + info!("Received delta for: {}", handle); + + }, + QueueEvent::ParentAdded(handle, _parent, _contact) => { + info!("Found parent for: {}", handle); +// if let Err(e) = caserver.update_entitlements(&handle) { +// error!("Error updating entitlements: {}", e); +// } + } + } + } + }); + scheduler.watch_thread(Duration::from_millis(100)) + } + pub fn service_base_uri(&self) -> &uri::Https { &self.service_uri } @@ -325,7 +379,7 @@ impl KrillServer { // Add publisher client for TA let req = admin::PublisherClientRequest::new( pub_handle, - PubServerInfo::for_krill( + PubServerContact::for_krill( self.service_uri.clone(), token ) @@ -401,7 +455,7 @@ impl KrillServer { // Add publisher for CA let req = admin::PublisherClientRequest::new( handle, - PubServerInfo::for_krill( + PubServerContact::for_krill( self.service_uri.clone(), token ) @@ -444,6 +498,15 @@ impl KrillServer { )?) } + pub fn rfc6492( + &self, + _parent: Handle, + _child: Handle, + _msg: SignedMessage + ) -> KrillRes { + unimplemented!("Got a message to debug") + } + } /// # Handle publication requests @@ -493,7 +556,7 @@ pub enum Error { SignerError(SignerError), #[display(fmt="{}", _0)] - CaServerError(CaServerError), + CaServerError(caserver::Error), #[display(fmt="{}", _0)] PubClientError(PubClientError), @@ -515,8 +578,8 @@ impl From for Error { fn from(e: SignerError) -> Self { Error::SignerError(e) } } -impl From> for Error { - fn from(e: CaServerError) -> Self { Error::CaServerError(e) } +impl From> for Error { + fn from(e: caserver::Error) -> Self { Error::CaServerError(e) } } impl From for Error { diff --git a/daemon/src/lib.rs b/daemon/src/lib.rs index a0b86f48..b8b6fe6a 100644 --- a/daemon/src/lib.rs +++ b/daemon/src/lib.rs @@ -25,12 +25,13 @@ extern crate toml; extern crate uuid; extern crate xml as xmlrs; -extern crate krill_ca; extern crate krill_commons; extern crate krill_pubc; extern crate krill_pubd; extern crate krill_client; + +pub mod ca; pub mod auth; pub mod config; pub mod endpoints; @@ -39,4 +40,6 @@ pub mod http; mod republisher; pub mod test; +mod mq; + diff --git a/daemon/src/mq.rs b/daemon/src/mq.rs new file mode 100644 index 00000000..9fcb814c --- /dev/null +++ b/daemon/src/mq.rs @@ -0,0 +1,124 @@ +//! A simple message queue, responsible for listening for (CA) events, +//! making them available for triggered processing, such as publishing +//! signed material, or asking a newly added parent for resource +//! entitlements. + +use std::collections::VecDeque; +use std::sync::RwLock; + +use krill_commons::api::admin::{ + Handle, + ParentCaContact +}; +use krill_commons::api::ca::PublicationDelta; +use krill_commons::eventsourcing::{ + Event, + EventListener, +}; + +use crate::ca::{ + CaSigner, + CaEvt, + CaEvtDet, + CertAuth, + ParentHandle +}; +use serde::export::fmt::Debug; + +//------------ QueueEvent ---------------------------------------------------- + +/// This type contains all the events of interest for a KrillServer, with +/// the details needed for triggered processing. +#[derive(Clone, Debug, Eq, PartialEq)] +pub enum QueueEvent { + ParentAdded(Handle, ParentHandle, ParentCaContact), + Delta(Handle, PublicationDelta) +} + +#[derive(Debug)] +pub struct EventQueueListener { + q: RwLock> +} + +impl EventQueueListener { + pub fn in_mem() -> Self { + EventQueueListener { q: RwLock::new(Box::new(MemoryEventQueue::new()))} + } +} + +impl EventQueueListener { + pub fn pop(&self) -> Option { + self.q.write().unwrap().pop() + } + + fn push_back(&self, evt: QueueEvent) { + self.q.write().unwrap().push_back(evt) + } +} + +// TODO: Is this unsafe here? I would think the RwLock is safe, but.. +unsafe impl Send for EventQueueListener {} +unsafe impl Sync for EventQueueListener {} + + +/// Implement listening for CertAuth Published events. +impl EventListener> for EventQueueListener { + fn listen(&self, _ca: &CertAuth, event: &CaEvt) { + let handle = event.handle(); + match event.details() { + CaEvtDet::Published(_,_,_, delta) | + CaEvtDet::TaPublished(delta) => { + let evt = QueueEvent::Delta(handle.clone(), delta.clone()); + self.push_back(evt); + }, + CaEvtDet::ParentAdded(parent, contact) => { + let evt = QueueEvent::ParentAdded( + handle.clone(), + parent.clone(), + contact.clone() + ); + self.push_back(evt); + } + _ => {} + } + } +} + +//------------ EventQueue ---------------------------------------------------- + +/// This trait provides the public contract for an EventQueue used by the +/// KrillServer. First implementation can be a simple in memory thing, but +/// we will need someting more robust, and possibly multi-master later. +/// +/// The EventQueue should implement Eventlistener +trait EventQueueStore: Debug { + fn pop(&self) -> Option; + fn push_back(&self, evt: QueueEvent); +} + + +//------------ MemoryEventQueue ---------------------------------------------- + +/// In memory event queue implementation. +#[derive(Debug)] +struct MemoryEventQueue { + q: RwLock> +} + +impl MemoryEventQueue { + pub fn new() -> Self { + MemoryEventQueue { q: RwLock::new(VecDeque::new())} + } +} + +impl EventQueueStore for MemoryEventQueue { + fn pop(&self) -> Option { + self.q.write().unwrap().pop_front() + } + + fn push_back(&self, evt: QueueEvent) { + self.q.write().unwrap().push_back(evt); + } +} + + diff --git a/daemon/tests/ca_under_ta.rs b/daemon/tests/ca_under_ta.rs index a6f55392..5f1542b4 100644 --- a/daemon/tests/ca_under_ta.rs +++ b/daemon/tests/ca_under_ta.rs @@ -2,9 +2,8 @@ extern crate krill_daemon; extern crate krill_client; extern crate krill_commons; extern crate krill_pubc; -extern crate krill_ca; -use krill_ca::ta_handle; +use krill_daemon::ca::caserver::ta_handle; use krill_client::options::{ CaCommand, Command, @@ -15,6 +14,8 @@ use krill_commons::api::ca::ResourceSet; use krill_commons::api::admin::{AddChildRequest, CertAuthInit, CertAuthPubMode, Handle, ParentCaContact, AddParentRequest, Token, ChildAuthRequest}; use krill_daemon::test::{ test_with_krill_server, execute_krillc_command }; use krill_commons::remote::rfc8183; +use std::thread; +use std::time::Duration; fn init_ta() {