eneros-gateway

eneros-gateway 是 EnerOS 所有控制指令的统一出入口，负责鉴权、约束二次校验、限流、审计记录与命令执行。任何对电网状态有副作用的操作（开关变位、出力调节、Agent 指令）都必须经过安全网关，未经网关的指令在内核态被直接拒绝。

职责描述

eneros-gateway 承担以下核心职责：

• 命令鉴权：基于角色（RBAC）与租户的多维度访问控制，结合 mTLS 客户端证书校验调用方身份
• 约束二次校验：调用 eneros-constraint 验证命令在当前电网状态下是否可行，越限即拒绝
• 双执行管线：实时域（< 1ms）走 RealtimeExecutor 处理保护/控制类指令；分析域走 CommandExecutor 处理规划/分析类指令
• 限流：按 IP / 租户 / Agent 维度的速率限制，防止恶意或失控 Agent 拖垮系统
• 审计落盘：所有命令经 eneros-audit HMAC 签名 + 链式哈希后落盘，符合 WORM（一次写入多次读取）要求
• 双因子确认：控制类指令需人工二次确认（require_two_factor）
• 事件广播：命令执行完成后通过 eneros-eventbus 广播事件，触发后续 Agent 与孪生体更新

关键类型与接口

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] 鉴权
        self.rbac.check(&cmd.agent_id, &cmd.action)?;

        // [2] 限流
        self.rate_limiter.check(&cmd.agent_id)?;

        // [3] 约束校验
        self.constraint_engine.check_command(&cmd)
            .map_err(GatewayError::ConstraintViolation)?;

        // [4] 双因子确认（控制类指令）
        if self.config.require_two_factor && cmd.action.is_control() {
            self.require_two_factor_confirmation(&cmd).await?;
        }

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

        // [6] 审计落盘
        if self.config.audit_enabled {
            self.audit.append(&cmd, &result).await?;
        }

        // [7] 事件广播
        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 / 错误

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),
}

核心 API

使用示例

基本命令执行

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!(
        "执行成功: status={:?} duration={:?}",
        result.status, result.duration
    ),
    Err(GatewayError::ConstraintViolation(v)) => {
        println!("约束拒绝: {}", v.message);
    }
    Err(GatewayError::Unauthorized) => println!("鉴权失败"),
    Err(GatewayError::RateLimited) => println!("限流拒绝"),
    Err(GatewayError::TwoFactorRequired) => println!("需要二次确认"),
    Err(e) => println!("其他错误: {:?}", e),
}

开关变位（实时域）

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!("开关变位结果: {:?}", result.status);

故障隔离与恢复

use eneros_gateway::{Command, Action};

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

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

DryRun 模式

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

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

// 仅校验不执行
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.status);

决策管线

命令进入
  ↓
[1] 鉴权 (RBAC + mTLS 客户端证书)
  ↓
[2] 限流检查（按 Agent / 租户 / IP 维度）
  ↓
[3] 约束校验 (eneros-constraint)
  ↓ 通过
[4] 双因子确认（控制类指令 require_two_factor）
  ↓ 确认
[5] 路由：实时域（< 1ms） / 分析域（< 1s）
  ↓
[6] 执行 + 审计落盘（HMAC 签名 + 链式哈希）
  ↓
[7] 事件广播 (eneros-eventbus)
  ↓
返回 CommandResult

配置参数表

GatewayConfig 字段

GatewayMode 模式

Action 路由表

性能指标

测试环境：4 核 / 8GB / Ubuntu 22.04 / Rust 1.78 release build。

依赖关系

自身依赖

[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"] }

被依赖

• eneros-api — API 层所有写操作经网关
• eneros-agent — Agent 指令经网关下发
• eneros-dashboard — 仪表盘控制操作经网关

版本与兼容性

相关文档

• 安全守卫 — 能力详解
• 双执行架构 — 实时/分析双域
• 限流与配额 — 限流策略
• eneros-constraint — 约束校验
• eneros-trust — mTLS 鉴权
• Crate 索引 — 完整列表