eneros-gateway

eneros-gateway is the unified entry point for all control commands in EnerOS, responsible for authentication, secondary constraint validation, rate limiting, audit logging, and command execution. Any operation with side effects on grid state (switch operations, output adjustment, Agent commands) must go through the safety gateway; commands bypassing the gateway are directly rejected in kernel mode.

Responsibilities

eneros-gateway handles the following core responsibilities:

• Command Authentication: Multi-dimensional access control based on roles (RBAC) and tenants, combined with mTLS client certificate verification of caller identity
• Secondary Constraint Validation: Calls eneros-constraint to verify whether the command is feasible under the current grid state; rejects if limits are exceeded
• Dual Execution Pipeline: Real-time domain (< 1ms) uses RealtimeExecutor for protection/control commands; analysis domain uses CommandExecutor for planning/analysis commands
• Rate Limiting: Rate limits per IP / tenant / Agent dimension, preventing malicious or runaway Agents from overwhelming the system
• Audit Logging: All commands are HMAC-signed + chained-hashed via eneros-audit before being persisted, meeting WORM (Write Once Read Many) requirements
• Two-Factor Confirmation: Control commands require manual secondary confirmation (require_two_factor)
• Event Broadcasting: After command execution completes, events are broadcast via eneros-eventbus, triggering subsequent Agent and Twin updates

Key Types and Interfaces

SafetyGateway

use eneros_constraint::ConstraintEngine;
use eneros_audit::AuditChain;
use eneros_trust::MtlsAcceptor;
use eneros_eventbus::EventBus;

pub struct SafetyGateway {
    constraint_engine: ConstraintEngine,
    audit: AuditChain,
    rate_limiter: RateLimiter,
    rbac: RbacEngine,
    realtime_executor: RealtimeExecutor,
    command_executor: CommandExecutor,
    config: GatewayConfig,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct GatewayConfig {
    pub mode: GatewayMode,
    pub rate_limit_per_second: u32,
    pub require_two_factor: bool,
    pub audit_enabled: bool,
    pub realtime_timeout_ms: u64,
    pub command_timeout_ms: u64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum GatewayMode {
    Enforce,
    Warn,
    DryRun,
}

impl Default for GatewayConfig {
    fn default() -> Self {
        Self {
            mode: GatewayMode::Enforce,
            rate_limit_per_second: 100,
            require_two_factor: true,
            audit_enabled: true,
            realtime_timeout_ms: 1,
            command_timeout_ms: 1000,
        }
    }
}

impl SafetyGateway {
    pub fn from_config(path: impl AsRef<Path>) -> Result<Self> {
        let text = std::fs::read_to_string(path.as_ref())?;
        let config: GatewayConfig = toml::from_str(&text)?;
        Ok(Self {
            constraint_engine: ConstraintEngine::default(),
            audit: AuditChain::new(),
            rate_limiter: RateLimiter::new(config.rate_limit_per_second),
            rbac: RbacEngine::default(),
            realtime_executor: RealtimeExecutor::new(),
            command_executor: CommandExecutor::new(),
            config,
        })
    }

    pub async fn execute(&self, cmd: Command) -> Result<CommandResult, GatewayError> {
        // [1] Authentication
        self.rbac.check(&cmd.agent_id, &cmd.action)?;

        // [2] Rate limiting
        self.rate_limiter.check(&cmd.agent_id)?;

        // [3] Constraint validation
        self.constraint_engine.check_command(&cmd)
            .map_err(GatewayError::ConstraintViolation)?;

        // [4] Two-factor confirmation (for control commands)
        if self.config.require_two_factor && cmd.action.is_control() {
            self.require_two_factor_confirmation(&cmd).await?;
        }

        // [5] Routing + execution
        let result = if cmd.action.is_realtime() {
            self.realtime_executor.execute(&cmd).await?
        } else {
            self.command_executor.execute(&cmd).await?
        };

        // [6] Audit logging
        if self.config.audit_enabled {
            self.audit.append(&cmd, &result).await?;
        }

        // [7] Event broadcast
        Ok(result)
    }
}

Command / Action

use eneros_core::{AgentId, BusId, BranchId, ElementId};
use chrono::{DateTime, Utc};

#[derive(Debug, Clone)]
pub struct Command {
    pub id: CommandId,
    pub agent_id: AgentId,
    pub action: Action,
    pub params: serde_json::Value,
    pub requested_at: DateTime<Utc>,
    pub tenant: Option<String>,
}

#[derive(Debug, Clone)]
pub struct CommandId(pub uuid::Uuid);

impl Command {
    pub fn new(agent_id: impl Into<String>) -> Self {
        Self {
            id: CommandId(uuid::Uuid::new_v4()),
            agent_id: AgentId::new(agent_id),
            action: Action::Noop,
            params: serde_json::Value::Null,
            requested_at: Utc::now(),
            tenant: None,
        }
    }

    pub fn action(mut self, action: Action) -> Self {
        self.action = action;
        self
    }

    pub fn params(mut self, params: serde_json::Value) -> Self {
        self.params = params;
        self
    }

    pub fn tenant(mut self, tenant: impl Into<String>) -> Self {
        self.tenant = Some(tenant.into());
        self
    }
}

#[derive(Debug, Clone)]
pub enum Action {
    Noop,
    SetGeneration { bus: BusId, mw: f64 },
    SetVoltage { bus: BusId, v_pu: f64 },
    SwitchBranch { branch: BranchId, status: BranchStatus },
    TapChange { transformer: ElementId, step: i32 },
    ShedLoad { bus: BusId, mw: f64 },
    IsolateFault { branch: BranchId },
    RestoreService { area: String },
}

impl Action {
    pub fn is_control(&self) -> bool {
        matches!(
            self,
            Action::SwitchBranch { .. }
                | Action::TapChange { .. }
                | Action::ShedLoad { .. }
                | Action::IsolateFault { .. }
                | Action::RestoreService { .. }
        )
    }

    pub fn is_realtime(&self) -> bool {
        matches!(
            self,
            Action::SwitchBranch { .. } | Action::IsolateFault { .. } | Action::ShedLoad { .. }
        )
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BranchStatus {
    Closed,
    Open,
}

CommandResult / Executor

#[derive(Debug, Clone)]
pub struct CommandResult {
    pub command_id: CommandId,
    pub status: CommandStatus,
    pub executed_at: DateTime<Utc>,
    pub duration: std::time::Duration,
    pub details: serde_json::Value,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CommandStatus {
    Success,
    Rejected,
    Failed,
    Timeout,
}

pub struct RealtimeExecutor {
    timeout: std::time::Duration,
}

impl RealtimeExecutor {
    pub fn new() -> Self {
        Self {
            timeout: std::time::Duration::from_millis(1),
        }
    }

    pub async fn execute(&self, cmd: &Command) -> Result<CommandResult, GatewayError> {
        let start = std::time::Instant::now();
        tokio::time::timeout(self.timeout, self.dispatch(cmd))
            .await
            .map_err(|_| GatewayError::Timeout)?
            .map(|_| CommandResult {
                command_id: cmd.id.clone(),
                status: CommandStatus::Success,
                executed_at: Utc::now(),
                duration: start.elapsed(),
                details: serde_json::json!({"executor": "realtime"}),
            })
    }

    async fn dispatch(&self, cmd: &Command) -> Result<(), GatewayError> {
        Ok(())
    }
}

pub struct CommandExecutor {
    timeout: std::time::Duration,
}

impl CommandExecutor {
    pub fn new() -> Self {
        Self {
            timeout: std::time::Duration::from_millis(1000),
        }
    }

    pub async fn execute(&self, cmd: &Command) -> Result<CommandResult, GatewayError> {
        let start = std::time::Instant::now();
        Ok(CommandResult {
            command_id: cmd.id.clone(),
            status: CommandStatus::Success,
            executed_at: Utc::now(),
            duration: start.elapsed(),
            details: serde_json::json!({"executor": "command"}),
        })
    }
}

RateLimiter / Errors

use std::collections::HashMap;
use std::time::{Duration, Instant};

pub struct RateLimiter {
    limit_per_second: u32,
    buckets: HashMap<AgentId, Vec<Instant>>,
}

impl RateLimiter {
    pub fn new(limit_per_second: u32) -> Self {
        Self {
            limit_per_second,
            buckets: HashMap::new(),
        }
    }

    pub fn check(&self, agent_id: &AgentId) -> Result<(), GatewayError> {
        let now = Instant::now();
        let window = Duration::from_secs(1);
        let bucket = self.buckets.get(agent_id);
        if let Some(history) = bucket {
            let recent: Vec<_> = history.iter().filter(|t| now.duration_since(**t) < window).collect();
            if recent.len() as u32 >= self.limit_per_second {
                return Err(GatewayError::RateLimited);
            }
        }
        Ok(())
    }
}

#[derive(Debug, Clone)]
pub enum GatewayError {
    Unauthorized,
    Forbidden(String),
    RateLimited,
    ConstraintViolation(eneros_constraint::Violation),
    Timeout,
    TwoFactorRequired,
    Internal(String),
}

Core API

Usage Examples

Basic Command Execution

use eneros_gateway::{SafetyGateway, Command, Action};

let gateway = SafetyGateway::from_config("gateway.toml")?;

let cmd = Command::new("agent_01")
    .action(Action::SetGeneration { bus: "bus_2".into(), mw: 50.0 });

match gateway.execute(cmd).await {
    Ok(result) => println!(
        "Execution success: status={:?} duration={:?}",
        result.status, result.duration
    ),
    Err(GatewayError::ConstraintViolation(v)) => {
        println!("Constraint rejected: {}", v.message);
    }
    Err(GatewayError::Unauthorized) => println!("Authentication failed"),
    Err(GatewayError::RateLimited) => println!("Rate limited"),
    Err(GatewayError::TwoFactorRequired) => println!("Two-factor confirmation required"),
    Err(e) => println!("Other error: {:?}", e),
}

Switch Operation (Real-time Domain)

use eneros_gateway::{Command, Action, BranchStatus};

let cmd = Command::new("selfheal_01")
    .action(Action::SwitchBranch {
        branch: "branch_5_8".into(),
        status: BranchStatus::Open,
    })
    .tenant("tenant_east");

let result = gateway.execute(cmd).await?;
println!("Switch operation result: {:?}", result.status);

Fault Isolation and Recovery

use eneros_gateway::{Command, Action};

// Fault isolation
let isolate_cmd = Command::new("selfheal_01")
    .action(Action::IsolateFault { branch: "branch_3_4".into() });
gateway.execute(isolate_cmd).await?;

// Service restoration
let restore_cmd = Command::new("selfheal_01")
    .action(Action::RestoreService { area: "east".into() });
gateway.execute(restore_cmd).await?;

DryRun Mode

use eneros_gateway::{SafetyGateway, GatewayConfig, GatewayMode};

let mut config = GatewayConfig::default();
config.mode = GatewayMode::DryRun;
let gateway = SafetyGateway::with_config(config);

// Validate only, no execution
let cmd = Command::new("planner_01")
    .action(Action::ShedLoad { bus: "bus_5".into(), mw: 10.0 });
let result = gateway.execute(cmd).await?;
println!("DryRun result: {:?}", result.status);

Decision Pipeline

Command entry
  ↓
[1] Authentication (RBAC + mTLS client certificate)
  ↓
[2] Rate limit check (per Agent / tenant / IP dimension)
  ↓
[3] Constraint validation (eneros-constraint)
  ↓ Passes
[4] Two-factor confirmation (control commands require_two_factor)
  ↓ Confirmed
[5] Routing: real-time domain (< 1ms) / analysis domain (< 1s)
  ↓
[6] Execution + audit logging (HMAC signature + chained hash)
  ↓
[7] Event broadcast (eneros-eventbus)
  ↓
Returns CommandResult

Configuration Parameter Table

GatewayConfig Fields

GatewayMode Modes

Action Routing Table

Performance Metrics

Test environment: 4 cores / 8GB / Ubuntu 22.04 / Rust 1.78 release build.

Dependencies

Own Dependencies

[dependencies]
eneros-core = { path = "../core", version = "0.47.0" }
eneros-constraint = { path = "../constraint", version = "0.47.0" }
eneros-trust = { path = "../trust", version = "0.47.0" }
eneros-audit = { path = "../audit", version = "0.47.0" }
eneros-eventbus = { path = "../eventbus", version = "0.47.0" }
tokio = { version = "1", features = ["full"] }
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
toml = "0.8"
tracing = "0.1"
uuid = { version = "1", features = ["v4", "serde"] }
chrono = { version = "0.4", features = ["serde"] }

Depended On By

• eneros-api — All write operations at the API layer go through the gateway
• eneros-agent — Agent commands are dispatched through the gateway
• eneros-dashboard — Dashboard control operations go through the gateway

Version and Compatibility

Related Documentation

• Safety Guard — Capability details
• Dual Execution Architecture — Real-time/analysis dual domains
• Rate Limiting and Quotas — Rate limiting policies
• eneros-constraint — Constraint validation
• eneros-trust — mTLS authentication
• Crate Index — Complete list