DaemonEye System Architecture#

#[async_trait]
pub trait ProcessCollector: Send + Sync {
    async fn enumerate_processes(&self) -> Result<Vec<ProcessRecord>>;
    async fn handle_detection_task(&self, task: DetectionTask) -> Result<DetectionResult>;
    async fn serve_ipc(&self) -> Result<()>;
}

#[async_trait]
pub trait HashComputer: Send + Sync {
    async fn compute_hash(&self, path: &Path) -> Result<Option<String>>;
    fn get_algorithm(&self) -> &'static str;
}

#[async_trait]
pub trait AuditLogger: Send + Sync {
    async fn log_event(&self, event: &AuditEvent) -> Result<()>;
    async fn verify_chain(&self) -> Result<ChainVerificationResult>;
}

pub struct ProcessCollector {
    config: CollectorConfig,
    hash_computer: Box<dyn HashComputer>,
    audit_logger: Box<dyn AuditLogger>,
    ipc_server: Box<dyn IpcServer>,
    privilege_manager: PrivilegeManager,
}

impl ProcessCollector {
    pub async fn new(config: CollectorConfig) -> Result<Self> {
        let mut privilege_manager = PrivilegeManager::new();

        // Request minimal required privileges
        privilege_manager.request_enhanced_privileges().await?;

        let collector = Self {
            config,
            hash_computer: Box::new(Sha256HashComputer::new()),
            audit_logger: Box::new(HashChainAuditLogger::new(&config.audit_path)?),
            ipc_server: Box::new(UnixSocketServer::new(&config.ipc_path)?),
            privilege_manager,
        };

        // Drop privileges immediately after initialization
        collector.privilege_manager.drop_privileges().await?;

        Ok(collector)
    }
}

#[async_trait]
pub trait DetectionEngine: Send + Sync {
    async fn execute_rules(&self, scan_id: i64) -> Result<Vec<Alert>>;
    async fn validate_sql(&self, query: &str) -> Result<ValidationResult>;
    async fn load_rules(&self) -> Result<Vec<DetectionRule>>;
}

#[async_trait]
pub trait AlertManager: Send + Sync {
    async fn generate_alert(&self, detection: DetectionResult) -> Result<Alert>;
    async fn deliver_alert(&self, alert: &Alert) -> Result<DeliveryResult>;
    async fn deduplicate_alert(&self, alert: &Alert) -> Result<Option<Alert>>;
}

#[async_trait]
pub trait ProcessManager: Send + Sync {
    async fn start_procmond(&self) -> Result<()>;
    async fn stop_procmond(&self) -> Result<()>;
    async fn restart_procmond(&self) -> Result<()>;
    async fn health_check(&self) -> Result<HealthStatus>;
}

pub struct DetectionEngine {
    // DataFusion session executing validated/derived SQL; redb per-domain
    // tables are exposed through TableProvider implementations (ADR-0006).
    query_ctx: datafusion::execution::context::SessionContext,
    table_providers: Vec<Arc<dyn datafusion::datasource::TableProvider>>,
    rule_manager: RuleManager,
    alert_manager: AlertManager,
    sql_validator: SqlValidator,
    ipc_client: IpcClient,
    process_manager: ProcessManager,
}

impl DetectionEngine {
    pub async fn new(config: AgentConfig) -> Result<Self> {
        // Build a locked-down DataFusion session and register redb-backed
        // TableProviders for each collector domain (process, etc.).
        let (query_ctx, table_providers) =
            build_session_context(&config.event_store_path, ALLOWED_SQL_FUNCTIONS)?;

        // Initialize SQL validator with security constraints
        let sql_validator = SqlValidator::new()
            .with_allowed_functions(ALLOWED_SQL_FUNCTIONS)
            .with_read_only_mode(true)
            .with_timeout(Duration::from_secs(30));

        // Initialize IPC client for procmond communication
        let ipc_client = IpcClient::new(&config.procmond_socket_path)?;

        // Start procmond process
        let process_manager = ProcessManager::new(config.procmond_config);
        process_manager.start_procmond().await?;

        Ok(Self {
            query_ctx,
            table_providers,
            rule_manager: RuleManager::new(&config.rules_path)?,
            alert_manager: AlertManager::new(&config.alerting_config)?,
            sql_validator,
            ipc_client,
            process_manager,
        })
    }
}

#[async_trait]
pub trait QueryExecutor: Send + Sync {
    async fn execute_query(&self, query: &str, params: &[Value]) -> Result<QueryResult>;
    async fn export_data(&self, format: ExportFormat, filter: &Filter) -> Result<ExportResult>;
}

#[async_trait]
pub trait HealthChecker: Send + Sync {
    async fn check_system_health(&self) -> Result<HealthStatus>;
    async fn check_component_health(&self, component: Component) -> Result<ComponentHealth>;
}

#[async_trait]
pub trait DataManager: Send + Sync {
    async fn export_alerts(
        &self,
        format: ExportFormat,
        filter: &AlertFilter,
    ) -> Result<ExportResult>;
    async fn export_processes(
        &self,
        format: ExportFormat,
        filter: &ProcessFilter,
    ) -> Result<ExportResult>;
}

pub mod config {
    pub mod environment;
    pub mod hierarchical;
    pub mod validation;
}

pub mod models {
    pub mod alert;
    pub mod audit;
    pub mod detection_rule;
    pub mod process;
}

pub mod storage {
    pub mod audit_ledger;
    pub mod connection_pool;
    pub mod redb;
}

pub mod detection {
    pub mod rule_engine;
    pub mod sandbox;
    pub mod sql_validator;
}

pub mod alerting {
    pub mod deduplication;
    pub mod delivery;
    pub mod sinks;
}

pub mod crypto {
    pub mod hash_chain;
    pub mod integrity;
    pub mod signatures;
}

pub mod telemetry {
    pub mod health;
    pub mod metrics;
    pub mod tracing;
}

// Hierarchical correlation tracking
let parent_metadata = CorrelationMetadata::new("workflow-id".to_string())
    .with_stage("detection".to_string())
    .with_tag("workflow".to_string(), "threat_analysis".to_string());

let child_metadata = parent_metadata.create_child("analysis-id".to_string());
// Child inherits workflow stage and tags from parent

syntax = "proto3";

// Simple detection tasks sent from daemoneye-agent to procmond
message DetectionTask {
    string task_id = 1;
    TaskType task_type = 2;
    optional ProcessFilter process_filter = 3;
    optional HashCheck hash_check = 4;
    optional string metadata = 5;
}

enum TaskType {
    ENUMERATE_PROCESSES = 0;
    CHECK_PROCESS_HASH = 1;
    MONITOR_PROCESS_TREE = 2;
    VERIFY_EXECUTABLE = 3;
}

message ProcessFilter {
    repeated string process_names = 1;
    repeated uint32 pids = 2;
    optional string executable_pattern = 3;
}

message HashCheck {
    string expected_hash = 1;
    string hash_algorithm = 2;
    string executable_path = 3;
}

// Results sent back from procmond to daemoneye-agent
message DetectionResult {
    string task_id = 1;
    bool success = 2;
    optional string error_message = 3;
    repeated ProcessRecord processes = 4;
    optional HashResult hash_result = 5;
}

message ProcessRecord {
    uint32 pid = 1;
    optional uint32 ppid = 2;
    string name = 3;
    optional string executable_path = 4;
    repeated string command_line = 5;
    optional int64 start_time = 6;
    optional double cpu_usage = 7;
    optional uint64 memory_usage = 8;
    optional string executable_hash = 9;
    optional string hash_algorithm = 10;
    optional string user_id = 11;
    bool accessible = 12;
    bool file_exists = 13;
    int64 collection_time = 14;
    // ssdeep/CTPH fuzzy hash (non-cryptographic, for binary-change detection).
    // Independent of executable_hash: may be absent while the SHA-256 identity
    // hash is present.
    optional string ssdeep_hash = 15;
    // On-disk-vs-running state: UNKNOWN / MATCH / MISMATCH, UNKNOWN = 0.
    // Three-state so an unprobed platform reports UNKNOWN rather than a
    // fabricated clean result; probed on Linux only today, so macOS and
    // Windows report UNKNOWN. UNKNOWN is never a clean result.
    OnDiskState on_disk_state = 16;
    // ssdeep computation failed while the SHA-256 identity hash succeeded —
    // degraded integrity coverage for this process.
    bool ssdeep_degraded = 17;
}

pub struct UnixSocketServer {
    path: PathBuf,
    listener: UnixListener,
}

impl IpcServer for UnixSocketServer {
    async fn serve<F>(&self, handler: F) -> Result<()>
    where
        F: Fn(DetectionTask) -> Result<DetectionResult> + Send + Sync + 'static,
    {
        let mut incoming = self.listener.incoming();

        while let Some(stream) = incoming.next().await {
            let stream = stream?;
            let handler = handler.clone();

            tokio::spawn(async move {
                Self::handle_connection(stream, handler).await;
            });
        }

        Ok(())
    }
}

pub struct NamedPipeServer {
    pipe_name: String,
    server: NamedPipeServerStream,
}

impl IpcServer for NamedPipeServer {
    async fn serve<F>(&self, handler: F) -> Result<()>
    where
        F: Fn(DetectionTask) -> Result<DetectionResult> + Send + Sync + 'static,
    {
        // Windows named pipe implementation
        // Similar to Unix socket but using Windows named pipes
    }
}

// Process snapshots table
pub struct ProcessSnapshot {
    pub id: Uuid,
    pub scan_id: i64,
    pub collection_time: i64,
    pub pid: u32,
    pub ppid: Option<u32>,
    pub name: String,
    pub executable_path: Option<PathBuf>,
    pub command_line: Vec<String>,
    pub start_time: Option<i64>,
    pub cpu_usage: Option<f64>,
    pub memory_usage: Option<u64>,
    pub executable_hash: Option<String>,
    pub hash_algorithm: Option<String>,
    pub user_id: Option<String>,
    pub accessible: bool,
    pub file_exists: bool,
    pub platform_data: Option<serde_json::Value>,
}

// Scan metadata table
pub struct ScanMetadata {
    pub id: i64,
    pub start_time: i64,
    pub end_time: i64,
    pub process_count: i32,
    pub collection_duration_ms: i64,
    pub system_info: SystemInfo,
}

// Detection rules table
pub struct DetectionRule {
    pub id: String,
    pub name: String,
    pub description: Option<String>,
    pub version: i32,
    pub sql_query: String,
    pub enabled: bool,
    pub severity: AlertSeverity,
    pub category: Option<String>,
    pub tags: Vec<String>,
    pub author: Option<String>,
    pub created_at: i64,
    pub updated_at: i64,
    pub source_type: RuleSourceType,
    pub source_path: Option<PathBuf>,
}

// Alerts table
pub struct Alert {
    pub id: Uuid,
    pub alert_time: i64,
    pub rule_id: String,
    pub title: String,
    pub description: String,
    pub severity: AlertSeverity,
    pub scan_id: Option<i64>,
    pub affected_processes: Vec<u32>,
    pub process_count: i32,
    pub alert_data: serde_json::Value,
    pub rule_execution_time_ms: Option<i64>,
    pub dedupe_key: String,
}

pub struct AuditChain {
    hasher: blake3::Hasher,
    signer: Option<ed25519_dalek::Keypair>,
    previous_hash: Option<blake3::Hash>,
}

impl AuditChain {
    pub fn append_entry(&mut self, entry: &AuditEntry) -> Result<AuditRecord> {
        let entry_data = serde_json::to_vec(entry)?;
        let entry_hash = blake3::hash(&entry_data);

        let record = AuditRecord {
            sequence: self.next_sequence(),
            timestamp: SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis() as i64,
            actor: entry.actor.clone(),
            action: entry.action.clone(),
            payload_hash: entry_hash,
            previous_hash: self.previous_hash,
            entry_hash: self.compute_entry_hash(&entry_hash)?,
            signature: self.sign_entry(&entry_hash)?,
        };

        self.previous_hash = Some(record.entry_hash);
        Ok(record)
    }

    pub fn verify_chain(&self, records: &[AuditRecord]) -> Result<VerificationResult> {
        // Verify hash chain integrity and signatures
        for (i, record) in records.iter().enumerate() {
            if i > 0 {
                let prev_record = &records[i - 1];
                if record.previous_hash != Some(prev_record.entry_hash) {
                    return Err(VerificationError::ChainBroken(i));
                }
            }

            // Verify entry hash
            let computed_hash = self.compute_entry_hash(&record.payload_hash)?;
            if record.entry_hash != computed_hash {
                return Err(VerificationError::HashMismatch(i));
            }

            // Verify signature if present
            if let Some(signature) = &record.signature {
                self.verify_signature(&record.payload_hash, signature)?;
            }
        }

        Ok(VerificationResult::Valid)
    }
}

pub struct PrivilegeManager {
    initial_privileges: Privileges,
    current_privileges: Privileges,
    drop_completed: bool,
}

impl PrivilegeManager {
    pub async fn request_enhanced_privileges(&mut self) -> Result<()> {
        // Platform-specific privilege escalation
        #[cfg(target_os = "linux")]
        self.request_linux_capabilities()?;

        #[cfg(target_os = "windows")]
        self.request_windows_privileges()?;

        #[cfg(target_os = "macos")]
        self.request_macos_entitlements()?;

        Ok(())
    }

    pub async fn drop_privileges(&mut self) -> Result<()> {
        // Immediate privilege drop after initialization
        self.drop_all_elevated_privileges()?;
        self.drop_completed = true;
        self.audit_privilege_drop().await?;
        Ok(())
    }
}

pub struct SqlValidator {
    parser: sqlparser::Parser<sqlparser::dialect::SQLiteDialect>,
    allowed_functions: HashSet<String>,
}

impl SqlValidator {
    pub fn validate_query(&self, sql: &str) -> Result<ValidationResult> {
        let ast = self.parser.parse_sql(sql)?;

        for statement in &ast {
            match statement {
                Statement::Query(query) => self.validate_select_query(query)?,
                _ => return Err(ValidationError::ForbiddenStatement),
            }
        }

        Ok(ValidationResult::Valid)
    }

    fn validate_select_query(&self, query: &Query) -> Result<()> {
        // Validate SELECT body, WHERE clauses, functions, etc.
        // Reject any non-whitelisted constructs
        self.validate_select_body(&query.body)?;
        self.validate_where_clause(&query.selection)?;
        Ok(())
    }
}

pub struct BoundedChannel<T> {
    sender: mpsc::Sender<T>,
    receiver: mpsc::Receiver<T>,
    capacity: usize,
    backpressure_policy: BackpressurePolicy,
}

impl<T> BoundedChannel<T> {
    pub async fn send(&self, item: T) -> Result<(), ChannelError> {
        match self.backpressure_policy {
            BackpressurePolicy::Block => {
                self.sender.send(item).await?;
            }
            BackpressurePolicy::Drop => {
                if self.sender.try_send(item).is_err() {
                    return Err(ChannelError::ChannelFull);
                }
            }
            BackpressurePolicy::Error => {
                self.sender.try_send(item)?;
            }
        }
        Ok(())
    }
}