WEBDIENSTE/NLnetLabs-krill
1
0
Fork 0
mirror of https://github.com/NLnetLabs/krill.git synced 2026-09-22 17:34:56 +02:00
Code Issues Packages Projects Releases Wiki Activity

Introducing message queue for asynchronous triggered processes: request certs, publish. (work in progress for #13)

Browse Source
This commit is contained in:
Tim Bruijnzeels
2019-07-19 21:20:18 +02:00
parent 446a1c339b
commit 38464aa2cf
18 changed files with 589 additions and 421 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.toml
+1 -1
View File
@@ -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" }
ca/Cargo.toml
-28
View File
@@ -1,28 +0,0 @@
[package]
name = "krill_ca"
version = "0.2.0"
authors = ["Tim Bruijnzeels <tim@nlnetlabs.nl>", "Martin Hoffmann <martin@nlnetlabs.nl>"]
[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" ]
ca/README.md
-7
View File
@@ -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.
ca/src/lib.rs
-273
View File
@@ -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::<CertAuth<OpenSslSigner>>::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();
})
}
}
commons/src/api/admin.rs
+23 -9
View File
@@ -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)
}
}
daemon/Cargo.toml
-3
View File
@@ -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"
daemon/defaults/krill.conf
+1 -1
View File
@@ -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
#
ca/src/ca.rs → daemon/src/ca/ca.rs
+10 -21
View File
@@ -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<Cert> {
let serial: Serial = rand::thread_rng().gen::<u128>().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<S> = SentCommand<CaCmdDet<S>>;
type ParentHandle = Handle;
pub type ParentHandle = Handle;
type ResourceClassName = String;
#[derive(Clone, Debug)]
@@ -647,7 +641,7 @@ impl<S: CaSigner> Aggregate for CertAuth<S> {
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<S: CaSigner> Aggregate for CertAuth<S> {
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<S: CaSigner> Aggregate for CertAuth<S> {
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();
ca/src/caserver.rs → daemon/src/ca/caserver.rs
+258 -21
View File
@@ -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<S: CaSigner> {
signer: Arc<RwLock<S>>,
ca_store: Arc<DiskAggregateStore<CertAuth<S>>>
@@ -47,10 +46,12 @@ impl<S: CaSigner> CaServer<S> {
/// initialised.
pub fn build(
work_dir: &PathBuf,
events_queue: Arc<EventQueueListener>,
pub_clients: Arc<PubClients>,
signer: S
) -> CaResult<Self, S> {
let mut ca_store = DiskAggregateStore::<CertAuth<S>>::new(work_dir, CA_NS)?;
ca_store.add_listener(events_queue);
ca_store.add_listener(pub_clients);
Ok(CaServer {
@@ -316,7 +317,8 @@ impl<S: CaSigner> CaServer<S> {
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<S: CaSigner> {
TrustAnchorNotInitialisedError,
#[display(fmt = "{}", _0)]
CaError(ca::Error),
CaError(CaError),
#[display(fmt = "CA {} was already initialised", _0)]
DuplicateCa(String),
@@ -511,8 +513,8 @@ impl<S: CaSigner> From<io::Error> for Error<S> {
fn from(e: io::Error) -> Self { Error::IoError(e) }
}
impl<S: CaSigner> From<ca::Error> for Error<S> {
fn from(e: ca::Error) -> Self { Error::CaError(e) }
impl<S: CaSigner> From<CaError> for Error<S> {
fn from(e: CaError) -> Self { Error::CaError(e) }
}
impl<S: CaSigner> From<AggregateStoreError> for Error<S> {
@@ -526,8 +528,41 @@ impl<S: CaSigner> From<AggregateStoreError> for Error<S> {
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::<OpenSslSigner>::build(&d, pub_clients, signer).unwrap();
let mut server = CaServer::<OpenSslSigner>::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::<CertAuth<OpenSslSigner>>::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();
})
}
}
daemon/src/ca/mod.rs
+21
View File
@@ -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;
ca/src/publishing.rs → daemon/src/ca/publishing.rs
+32 -30
View File
@@ -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<PubClientInitDetails>;
#[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")
}
}
}
ca/src/signing.rs → daemon/src/ca/signing.rs
View File
daemon/src/config.rs
+5 -3
View File
@@ -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.
daemon/src/endpoints.rs
+34 -11
View File
@@ -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<Handle>
) -> 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<Handle>
) -> 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<AppServer>,
auth: Auth,
parent: Path<Handle>,
child: Path<Handle>,
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<OpenSslSigner> {
impl ErrorToStatus for caserver::Error<OpenSslSigner> {
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<OpenSslSigner> {
impl ToErrorCode for caserver::Error<OpenSslSigner> {
fn code(&self) -> ErrorCode {
match self {
CaServerError::CaError(e) => e.code(),
caserver::Error::CaError(e) => e.code(),
_ => ErrorCode::CaServerError
}
}
daemon/src/krillserver.rs
+73 -10
View File
@@ -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<EventQueueListener>,
_caserver: CaServer<OpenSslSigner>
) -> 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<Captured> {
unimplemented!("Got a message to debug")
}
}
/// # Handle publication requests
@@ -493,7 +556,7 @@ pub enum Error {
SignerError(SignerError),
#[display(fmt="{}", _0)]
CaServerError(CaServerError<OpenSslSigner>),
CaServerError(caserver::Error<OpenSslSigner>),
#[display(fmt="{}", _0)]
PubClientError(PubClientError),
@@ -515,8 +578,8 @@ impl From<SignerError> for Error {
fn from(e: SignerError) -> Self { Error::SignerError(e) }
}
impl From<CaServerError<OpenSslSigner>> for Error {
fn from(e: CaServerError<OpenSslSigner>) -> Self { Error::CaServerError(e) }
impl From<caserver::Error<OpenSslSigner>> for Error {
fn from(e: caserver::Error<OpenSslSigner>) -> Self { Error::CaServerError(e) }
}
impl From<PubClientError> for Error {
daemon/src/lib.rs
+4 -1
View File
@@ -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;
daemon/src/mq.rs
+124
View File
@@ -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<Box<EventQueueStore>>
}
impl EventQueueListener {
pub fn in_mem() -> Self {
EventQueueListener { q: RwLock::new(Box::new(MemoryEventQueue::new()))}
}
}
impl EventQueueListener {
pub fn pop(&self) -> Option<QueueEvent> {
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<S: CaSigner> EventListener<CertAuth<S>> for EventQueueListener {
fn listen(&self, _ca: &CertAuth<S>, 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<QueueEvent>;
fn push_back(&self, evt: QueueEvent);
}
//------------ MemoryEventQueue ----------------------------------------------
/// In memory event queue implementation.
#[derive(Debug)]
struct MemoryEventQueue {
q: RwLock<VecDeque<QueueEvent>>
}
impl MemoryEventQueue {
pub fn new() -> Self {
MemoryEventQueue { q: RwLock::new(VecDeque::new())}
}
}
impl EventQueueStore for MemoryEventQueue {
fn pop(&self) -> Option<QueueEvent> {
self.q.write().unwrap().pop_front()
}
fn push_back(&self, evt: QueueEvent) {
self.q.write().unwrap().push_back(evt);
}
}
daemon/tests/ca_under_ta.rs
+3 -2
View File
@@ -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() {