Implement giganto cluster - publish message

.gitattributes

@@ -0,0 +1 @@
+*.pem binary

CHANGELOG.md

@@ -5,7 +5,7 @@ file is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and
 this project adheres to [Semantic
 Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [Unreleased]
+## [0.16.0] - 2024-01-08
 
 ### Added
 
@@ -19,14 +19,10 @@ Versioning](https://semver.org/spec/v2.0.0.html).
   - Added `set_ack_transmission_count` GraphQL query to set the ack transmission
     count.This query changes the `AckTransmissionCount` used in ingest and
     `ack_transmission` in the config file to the input `count` value.
-- Added giganto cluster support for GraphQL requests.
 - Added documentation for implementing cluster-supported GraphQL APIs in
   `docs/guide-giganto-cluster-graphql.md`.
 - Added `ConvertGraphQLEdgesNode` derive macro that implements `From` trait from
   GraphQL client structs to project structs.
-- Added `request_from_peer: Option<bool>` argument to GraphQL endpoints:
-  `netflow5_raw_events`, `netflow9_raw_events`, `secu_log_raw_events`,
-  `statistics`.
 - Supported `log-broker` to send/receive operation log with redis server.
   - Set the redis server with `redis_log_address`, `redis_log_agent_id` and
     `redis_log_fetch_interval` in configuration options.
@@ -42,8 +38,13 @@ Versioning](https://semver.org/spec/v2.0.0.html).
   mode can also be read correctly.
 - Changed `export` GraphQL query's response value format from `{export_path}` to
 `{export_path}@{giganto_node_name}`
-- Changed `PEER_VERSION_REQ` to ">=0.16.0-alpha.1,<0.17.0"
 - Changed logging from `tracing` to `log-broker`.
+- Changed `PEER_VERSION_REQ` to ">=0.16.0,<0.17.0"
+- Changed `PUBLISH_VERSION_REQ` to ">=0.16.0,<0.17.0"
+- Added giganto cluster support for GraphQL and publish message requests.
+- Added `request_from_peer: Option<bool>` argument to GraphQL endpoints:
+  `netflow5_raw_events`, `netflow9_raw_events`, `secu_log_raw_events`,
+  `statistics`.
 
 ### Fixed
 
@@ -400,7 +401,7 @@ Versioning](https://semver.org/spec/v2.0.0.html).
 
 - Initial release.
 
-[Unreleased]: <https://github.com/aicers/giganto/compare/0.15.4...main>
+[0.16.0]: <https://github.com/aicers/giganto/compare/0.15.3...0.16.0>
 [0.15.4]: <https://github.com/aicers/giganto/compare/0.15.3...0.15.4>
 [0.15.3]: <https://github.com/aicers/giganto/compare/0.15.2...0.15.3>
 [0.15.2]: <https://github.com/aicers/giganto/compare/0.15.1...0.15.2>

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "giganto"
-version = "0.16.0-alpha.1"
+version = "0.16.0"
 edition = "2021"
 
 [lib]
@@ -52,10 +52,11 @@ warp = { version = "0.3", features = ["tls"] }
 x509-parser = "0.15"
 
 [dev-dependencies]
+mockito = "1.2.0"
+regex = "1"
+serial_test = "2.0.0"
 tempfile = "3"
 url = "2"
-regex = "1"
-mockito = "1.2.0"
 
 [features]
 default = ["benchmark"]

src/graphql/packet.rs

@@ -71,6 +71,7 @@ impl RawEventFilter for PacketFilter {
 #[allow(clippy::struct_field_names)]
 #[derive(SimpleObject, ConvertGraphQLEdgesNode)]
 #[graphql_client_type(names = [packets::PacketsPacketsEdgesNode, ])]
+#[allow(clippy::struct_field_names)]
 struct Packet {
     request_time: DateTime<Utc>,
     packet_time: DateTime<Utc>,

src/ingest.rs

@@ -4,7 +4,7 @@ mod tests;
 
 use crate::publish::send_direct_stream;
 use crate::server::{
-    certificate_info, config_server, extract_cert_from_conn, SERVER_CONNNECTION_DELAY,
+    certificate_info, config_server, extract_cert_from_conn, Certs, SERVER_CONNNECTION_DELAY,
     SERVER_ENDPOINT_DELAY,
 };
 use crate::storage::{Database, RawEventStore, StorageKey};
@@ -27,7 +27,6 @@ use giganto_client::{
 };
 use log_broker::{error, info, LogLocation};
 use quinn::{Endpoint, RecvStream, SendStream, ServerConfig};
-use rustls::{Certificate, PrivateKey};
 use std::sync::atomic::AtomicU16;
 use std::{
     net::SocketAddr,
@@ -68,14 +67,9 @@ pub struct Server {
 }
 
 impl Server {
-    pub fn new(
-        addr: SocketAddr,
-        certs: Vec<Certificate>,
-        key: PrivateKey,
-        files: Vec<Vec<u8>>,
-    ) -> Self {
-        let server_config = config_server(certs, key, files)
-            .expect("server configuration error with cert, key or root");
+    pub fn new(addr: SocketAddr, certs: &Arc<Certs>) -> Self {
+        let server_config =
+            config_server(certs).expect("server configuration error with cert, key or root");
         Server {
             server_config,
             server_address: addr,

src/ingest/tests.rs

@@ -2,7 +2,7 @@ use super::Server;
 use crate::{
     new_ingest_sources, new_pcap_sources, new_stream_direct_channels,
     storage::{Database, DbOptions},
-    to_cert_chain, to_private_key,
+    to_cert_chain, to_private_key, to_root_cert, Certs,
 };
 use base64::{engine::general_purpose::STANDARD as base64_engine, Engine};
 use chrono::{Duration, Utc};
@@ -22,6 +22,7 @@ use giganto_client::{
     RawEventKind,
 };
 use quinn::{Connection, Endpoint};
+use std::path::PathBuf;
 use std::{
     fs,
     net::{IpAddr, Ipv6Addr, SocketAddr},
@@ -40,10 +41,10 @@ fn get_token() -> &'static Mutex<u32> {
     TOKEN.get_or_init(|| Mutex::new(0))
 }
 
-const CERT_PATH: &str = "tests/cert.pem";
-const KEY_PATH: &str = "tests/key.pem";
-const CA_CERT_PATH: &str = "tests/root.pem";
-const HOST: &str = "localhost";
+const CERT_PATH: &str = "tests/certs/node1/cert.pem";
+const KEY_PATH: &str = "tests/certs/node1/key.pem";
+const CA_CERT_PATH: &str = "tests/certs/root.pem";
+const HOST: &str = "node1";
 const TEST_PORT: u16 = 60190;
 const PROTOCOL_VERSION: &str = "0.15.2";
 
@@ -75,13 +76,18 @@ fn server() -> Server {
     let cert = to_cert_chain(&cert_pem).unwrap();
     let key_pem = fs::read(KEY_PATH).unwrap();
     let key = to_private_key(&key_pem).unwrap();
-    let ca_cert = fs::read("tests/root.pem").unwrap();
+    let ca_cert_path: Vec<PathBuf> = vec![PathBuf::from(CA_CERT_PATH)];
+    let ca_certs = to_root_cert(&ca_cert_path).unwrap();
+
+    let certs = Arc::new(Certs {
+        certs: cert,
+        key,
+        ca_certs,
+    });
 
     Server::new(
         SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), TEST_PORT),
-        cert,
-        key,
-        vec![ca_cert],
+        &certs,
     )
 }