feat(db): rough database implementation

.env.example

@@ -4,5 +4,7 @@ DTIM__DEFAULT__PORT=3030
 DTIM__DEFAULT__LOG_LEVEL="info"
 
 # Sensitive info (.env only)
-DTIM__DEFAULT__STORAGE__DATABASE_URL="postgres://user:pass@localhost/db"
-DTIM__DEFAULT__WATCHERS__VIRUSTOTAL_API_KEY="your_api_key"
+DATABASE_URL="postgres://postgres:@localhost:5432/postgres" # For Diesel CLI
+
+DTIM__DEFAULT__STORAGE__DATABASE_URL=${DATABASE_URL}
+DTIM__DEFAULT__WATCHERS__VIRUSTOTAL_API_KEY="your_api_key"

Cargo.toml

@@ -8,7 +8,7 @@ default-run = "dtim"
 axum = { version = "0.8", features = ["http2", "tokio"] }
 axum-server = { version = "0.7", features = ["tls-rustls"] }
 chrono = { version = "0.4", features = ["serde"] }
-uuid = { version = "1.16", features = ["v7", "serde"] }
+uuid = { version = "1.16", features = ["v5", "v7", "serde"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "1.0", features = ["preserve_order"] }
 serde_with = "3.12"
@@ -28,3 +28,4 @@ async-trait = "0.1"
 http-body-util = "0.1"
 dotenvy = "0.15"
 once_cell = "1.21"
+diesel = { version = "2.2.0", features = ["postgres", "r2d2", "chrono"] }

diesel.toml

@@ -0,0 +1,9 @@
+# For documentation on how to configure this file,
+# see https://diesel.rs/guides/configuring-diesel-cli
+
+[print_schema]
+file = "src/db/schema.rs"
+custom_type_derives = ["diesel::query_builder::QueryId", "Clone"]
+
+[migrations_directory]
+dir = "/Users/caden/Developer/dtim/migrations"

migrations/00000000000000_diesel_initial_setup/down.sql

@@ -0,0 +1,6 @@
+-- This file was automatically created by Diesel to setup helper functions
+-- and other internal bookkeeping. This file is safe to edit, any future
+-- changes will be added to existing projects as new migrations.
+
+DROP FUNCTION IF EXISTS diesel_manage_updated_at(_tbl regclass);
+DROP FUNCTION IF EXISTS diesel_set_updated_at();

migrations/00000000000000_diesel_initial_setup/up.sql

@@ -0,0 +1,36 @@
+-- This file was automatically created by Diesel to setup helper functions
+-- and other internal bookkeeping. This file is safe to edit, any future
+-- changes will be added to existing projects as new migrations.
+
+
+
+
+-- Sets up a trigger for the given table to automatically set a column called
+-- `updated_at` whenever the row is modified (unless `updated_at` was included
+-- in the modified columns)
+--
+-- # Example
+--
+-- ```sql
+-- CREATE TABLE users (id SERIAL PRIMARY KEY, updated_at TIMESTAMP NOT NULL DEFAULT NOW());
+--
+-- SELECT diesel_manage_updated_at('users');
+-- ```
+CREATE OR REPLACE FUNCTION diesel_manage_updated_at(_tbl regclass) RETURNS VOID AS $$
+BEGIN
+    EXECUTE format('CREATE TRIGGER set_updated_at BEFORE UPDATE ON %s
+                    FOR EACH ROW EXECUTE PROCEDURE diesel_set_updated_at()', _tbl);
+END;
+$$ LANGUAGE plpgsql;
+
+CREATE OR REPLACE FUNCTION diesel_set_updated_at() RETURNS trigger AS $$
+BEGIN
+    IF (
+        NEW IS DISTINCT FROM OLD AND
+        NEW.updated_at IS NOT DISTINCT FROM OLD.updated_at
+    ) THEN
+        NEW.updated_at := current_timestamp;
+    END IF;
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;

migrations/2025-05-21-234457_create_encrypted_indicators/down.sql

@@ -0,0 +1 @@
+DROP TABLE encrypted_indicators;

migrations/2025-05-21-234457_create_encrypted_indicators/up.sql

@@ -0,0 +1,7 @@
+CREATE TABLE encrypted_indicators (
+    id CHAR(64) PRIMARY KEY,
+    ciphertext BYTEA NOT NULL,
+    nonce BYTEA NOT NULL,
+    mac BYTEA NOT NULL,
+    tlp_level TEXT NOT NULL
+);

src/api.rs

@@ -13,19 +13,17 @@ use ed25519_dalek::VerifyingKey;
 use http_body_util::BodyExt as _;
 use rustls::ServerConfig;
 use serde::Serialize;
-use std::{str::FromStr as _, sync::Arc};
+use std::sync::Arc;
 use tokio::sync::Mutex;
 
+use crate::models::ThreatIndicator;
 use crate::{
     crypto::MeshIdentity,
+    db::models::EncryptedIndicator,
     error::{ApiError, ApiErrorResponse},
     node::{Node, NodePeer},
 };
 use crate::{crypto::SymmetricKeyManager, models::StixBundle};
-use crate::{
-    models::{EncryptedThreatIndicator, ThreatIndicator},
-    uuid::Uuid,
-};
 
 #[derive(Clone)]
 pub struct AppState {
@@ -200,13 +198,14 @@ struct GossipIndicatorsResponse {
 
 async fn gossip_indicators_handler(
     State(state): State<Arc<AppState>>,
-    Json(indicators): Json<Vec<EncryptedThreatIndicator>>,
+    Json(indicators): Json<Vec<EncryptedIndicator>>,
 ) -> ApiResponse<GossipIndicatorsResponse> {
     let mut node = state.node.lock().await;
     let mut count = 0;
     for encrypted in indicators {
         if let Ok(indicator) = ThreatIndicator::decrypt(&encrypted, &state.key_mgr) {
-            node.add_or_increment_indicator(indicator);
+            node.add_or_increment_indicator(indicator)
+                .map_err(|_| ApiError::INTERNAL_SERVER_ERROR)?;
             count += 1;
         }
     }
@@ -229,7 +228,9 @@ async fn get_public_indicators_handler(
     State(state): State<Arc<AppState>>,
 ) -> ApiResponse<GetIndicatorsResponse> {
     let node = state.node.lock().await;
-    let indicators = node.list_indicators_by_tlp(crate::models::TlpLevel::White);
+    let indicators = node
+        .list_indicators_by_tlp(crate::models::TlpLevel::White)
+        .map_err(|_| ApiError::INTERNAL_SERVER_ERROR)?;
     // Return anonymized (open) view
     Ok((
         StatusCode::OK,
@@ -244,7 +245,9 @@ async fn get_private_indicators_handler(
     State(state): State<Arc<AppState>>,
 ) -> ApiResponse<GetIndicatorsResponse> {
     let node = state.node.lock().await;
-    let indicators = node.list_indicators_by_tlp(crate::models::TlpLevel::Red);
+    let indicators = node
+        .list_indicators_by_tlp(crate::models::TlpLevel::Red)
+        .map_err(|_| ApiError::INTERNAL_SERVER_ERROR)?;
     // TODO: Ensure the node is an authenticated recipient for private indicators
     // Return anonymized (moderate) view
     Ok((
@@ -261,8 +264,9 @@ async fn get_indicator_by_id_handler(
     Path(id): Path<String>,
 ) -> ApiResponse<serde_json::Value> {
     let node = state.node.lock().await;
-    let id = Uuid::from_str(&id).map_err(|_| ApiError::INVALID_INDICATOR_ID)?;
-    let indicator = node.get_indicator_by_id(&id).ok_or(ApiError::NOT_FOUND)?;
+    let indicator = node
+        .get_indicator_by_id(&id)
+        .map_err(|_| ApiError::NOT_FOUND)?;
     Ok((StatusCode::OK, Json(indicator.to_json(node.get_level()))))
 }
 
@@ -319,7 +323,12 @@ async fn taxii_get_objects_handler(
     Path(_collection_id): Path<String>,
 ) -> ApiResponse<serde_json::Value> {
     let node = state.node.lock().await;
-    let stix_objects = node.list_objects_by_tlp(crate::models::TlpLevel::White);
+    let stix_objects = node
+        .list_objects_by_tlp(crate::models::TlpLevel::White)
+        .map_err(|error| {
+            println!("Error: {:?}", error);
+            ApiError::INTERNAL_SERVER_ERROR
+        })?;
     let bundle = StixBundle::new(stix_objects);
     Ok((StatusCode::OK, Json(bundle.to_stix())))
 }
@@ -331,6 +340,7 @@ async fn taxii_post_objects_handler(
 ) -> ApiResponse<serde_json::Value> {
     let indicator = ThreatIndicator::from_stix(stix).map_err(|_| ApiError::INVALID_STIX_OBJECT)?;
     let mut node = state.node.lock().await;
-    node.add_indicator(indicator);
+    node.add_indicator(indicator)
+        .map_err(|_| ApiError::INTERNAL_SERVER_ERROR)?;
     Ok((StatusCode::OK, Json(serde_json::json!({ "status": "ok" }))))
 }