feat(os): add cognitive control execution bridge

This commit is contained in:
冰朔 2026-08-07 13:45:27 +08:00
commit 25959f7af1
11 changed files with 877 additions and 6 deletions

View file

@ -0,0 +1,10 @@
[package]
name = "guanghu-execution-bridge"
version = "0.1.0"
edition = "2021"
license = "AGPL-3.0-or-later"
description = "Deterministic bridge from Guanghu protocol decisions to allowlisted Linux execution adapters"
[dependencies]
serde = { version = "1", features = ["derive"] }
serde_json = "1"

View file

@ -0,0 +1,357 @@
use std::{
fmt,
process::{Command, Output},
};
use serde::{Deserialize, Serialize};
pub const REQUEST_SCHEMA: &str = "guanghu.execution-request/v1";
pub const POLICY_SCHEMA: &str = "guanghu.execution-policy/v1";
pub const REQUIRED_PROTOCOL_CHAIN: [&str; 10] = [
"GLS-0301", // message envelope
"GLS-0302", // identity
"GLS-0303", // context
"GLS-0306", // receipt
"GLS-0309", // work order
"GLS-0311", // witness
"GLS-0130", // compiler
"GLS-0131", // deterministic representation
"GLS-0709", // adapter
"GLS-0710", // immutable module
];
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
pub struct ExecutionRequest {
pub schema: String,
pub request_id: String,
pub subject_id: String,
pub target_node_id: String,
pub protocol_chain: Vec<String>,
pub action: ExecutionAction,
pub authorization: Option<AuthorizationReference>,
pub rollback: Option<RollbackReference>,
}
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
pub struct ExecutionAction {
pub kind: ActionKind,
pub resource: String,
}
#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ActionKind {
ServiceStatus,
ServiceRestart,
}
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
pub struct AuthorizationReference {
pub authorization_id: String,
pub allowed_action: ActionKind,
pub target_node_id: String,
}
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
pub struct RollbackReference {
pub checkpoint_id: String,
pub recovery_action: String,
}
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
pub struct ExecutionPolicy {
pub schema: String,
pub policy_id: String,
pub target_node_id: String,
pub allowed_services: Vec<String>,
pub allow_status: bool,
pub allow_restart: bool,
}
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
pub struct ExecutionPlan {
pub request_id: String,
pub subject_id: String,
pub target_node_id: String,
pub policy_id: String,
pub protocol_chain: Vec<String>,
pub action: ExecutionAction,
pub adapter: String,
pub program: String,
pub arguments: Vec<String>,
pub mutating: bool,
pub authorization_id: Option<String>,
pub rollback_checkpoint_id: Option<String>,
}
#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
pub struct ExecutionReceipt {
pub schema: String,
pub request_id: String,
pub subject_id: String,
pub target_node_id: String,
pub policy_id: String,
pub action: ExecutionAction,
pub adapter: String,
pub accepted: bool,
pub command_exit_code: Option<i32>,
pub target_state_verified: bool,
pub final_state: String,
pub stdout: String,
pub stderr: String,
pub rollback_checkpoint_id: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct CommandResult {
pub exit_code: Option<i32>,
pub stdout: String,
pub stderr: String,
}
impl From<Output> for CommandResult {
fn from(output: Output) -> Self {
Self {
exit_code: output.status.code(),
stdout: String::from_utf8_lossy(&output.stdout).trim().to_owned(),
stderr: String::from_utf8_lossy(&output.stderr).trim().to_owned(),
}
}
}
pub trait CommandExecutor {
fn execute(&self, program: &str, arguments: &[String]) -> Result<CommandResult, String>;
}
pub struct LinuxCommandExecutor;
impl CommandExecutor for LinuxCommandExecutor {
fn execute(&self, program: &str, arguments: &[String]) -> Result<CommandResult, String> {
Command::new(program)
.args(arguments)
.output()
.map(CommandResult::from)
.map_err(|error| format!("cannot execute allowlisted adapter: {error}"))
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct BridgeError(String);
impl BridgeError {
fn new(message: impl Into<String>) -> Self {
Self(message.into())
}
}
impl fmt::Display for BridgeError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.0)
}
}
impl std::error::Error for BridgeError {}
pub fn compile_request(
request: &ExecutionRequest,
policy: &ExecutionPolicy,
) -> Result<ExecutionPlan, BridgeError> {
require(
request.schema == REQUEST_SCHEMA,
"unsupported execution request schema",
)?;
require(
policy.schema == POLICY_SCHEMA,
"unsupported execution policy schema",
)?;
require(
!request.request_id.trim().is_empty(),
"request_id is required",
)?;
require(
!request.subject_id.trim().is_empty(),
"subject_id is required",
)?;
require(
request.target_node_id == policy.target_node_id,
"request target does not match policy target",
)?;
for required in REQUIRED_PROTOCOL_CHAIN {
require(
request.protocol_chain.iter().any(|id| id == required),
format!("required protocol is missing: {required}"),
)?;
}
require(
policy
.allowed_services
.iter()
.any(|service| service == &request.action.resource),
"service is not allowlisted by the execution policy",
)?;
let (arguments, mutating) = match request.action.kind {
ActionKind::ServiceStatus => {
require(policy.allow_status, "service status is disabled by policy")?;
(
vec!["is-active".to_owned(), request.action.resource.clone()],
false,
)
}
ActionKind::ServiceRestart => {
require(
policy.allow_restart,
"service restart is disabled by policy",
)?;
validate_mutation_references(request)?;
(
vec!["restart".to_owned(), request.action.resource.clone()],
true,
)
}
};
Ok(ExecutionPlan {
request_id: request.request_id.clone(),
subject_id: request.subject_id.clone(),
target_node_id: request.target_node_id.clone(),
policy_id: policy.policy_id.clone(),
protocol_chain: request.protocol_chain.clone(),
action: request.action.clone(),
adapter: "LINUX_SYSTEMD_V1".to_owned(),
program: "/usr/bin/systemctl".to_owned(),
arguments,
mutating,
authorization_id: request
.authorization
.as_ref()
.map(|authorization| authorization.authorization_id.clone()),
rollback_checkpoint_id: request
.rollback
.as_ref()
.map(|rollback| rollback.checkpoint_id.clone()),
})
}
fn validate_mutation_references(request: &ExecutionRequest) -> Result<(), BridgeError> {
let authorization = request
.authorization
.as_ref()
.ok_or_else(|| BridgeError::new("mutating action requires authorization"))?;
require(
!authorization.authorization_id.trim().is_empty(),
"authorization_id is required",
)?;
require(
authorization.allowed_action == request.action.kind,
"authorization action does not match request action",
)?;
require(
authorization.target_node_id == request.target_node_id,
"authorization target does not match request target",
)?;
let rollback = request
.rollback
.as_ref()
.ok_or_else(|| BridgeError::new("mutating action requires rollback reference"))?;
require(
!rollback.checkpoint_id.trim().is_empty(),
"rollback checkpoint_id is required",
)?;
require(
!rollback.recovery_action.trim().is_empty(),
"rollback recovery_action is required",
)
}
pub fn execute_plan(
plan: &ExecutionPlan,
executor: &dyn CommandExecutor,
) -> Result<ExecutionReceipt, BridgeError> {
let expected_operation = match plan.action.kind {
ActionKind::ServiceStatus => "is-active",
ActionKind::ServiceRestart => "restart",
};
require(
plan.adapter == "LINUX_SYSTEMD_V1"
&& plan.program == "/usr/bin/systemctl"
&& matches!(
plan.arguments.as_slice(),
[operation, service]
if operation == expected_operation
&& service == &plan.action.resource
),
"compiled plan is not an allowlisted Linux systemd adapter",
)?;
let result = executor
.execute(&plan.program, &plan.arguments)
.map_err(BridgeError::new)?;
let command_succeeded = result.exit_code == Some(0);
let (target_state_verified, final_state, stdout, stderr, exit_code) = if plan.action.kind
== ActionKind::ServiceRestart
&& command_succeeded
{
let verification_arguments = vec!["is-active".to_owned(), plan.action.resource.clone()];
let verification = executor
.execute(&plan.program, &verification_arguments)
.map_err(BridgeError::new)?;
let verified = verification.exit_code == Some(0) && verification.stdout.trim() == "active";
(
verified,
if verified { "PASS_100" } else { "FAIL_0" }.to_owned(),
join_observations(&result.stdout, &verification.stdout),
join_observations(&result.stderr, &verification.stderr),
verification.exit_code,
)
} else {
let verified = command_succeeded
&& (plan.action.kind != ActionKind::ServiceStatus || result.stdout.trim() == "active");
(
verified,
if verified { "PASS_100" } else { "FAIL_0" }.to_owned(),
result.stdout,
result.stderr,
result.exit_code,
)
};
Ok(ExecutionReceipt {
schema: "guanghu.execution-receipt/v1".to_owned(),
request_id: plan.request_id.clone(),
subject_id: plan.subject_id.clone(),
target_node_id: plan.target_node_id.clone(),
policy_id: plan.policy_id.clone(),
action: plan.action.clone(),
adapter: plan.adapter.clone(),
accepted: true,
command_exit_code: exit_code,
target_state_verified,
final_state,
stdout,
stderr,
rollback_checkpoint_id: plan.rollback_checkpoint_id.clone(),
})
}
fn require(condition: bool, message: impl Into<String>) -> Result<(), BridgeError> {
if condition {
Ok(())
} else {
Err(BridgeError::new(message))
}
}
fn join_observations(first: &str, second: &str) -> String {
match (first.is_empty(), second.is_empty()) {
(true, true) => String::new(),
(false, true) => first.to_owned(),
(true, false) => second.to_owned(),
(false, false) => format!("{first}\n{second}"),
}
}

View file

@ -0,0 +1,61 @@
use std::{env, fs, process::ExitCode};
use guanghu_execution_bridge::{
compile_request, execute_plan, ExecutionPolicy, ExecutionRequest, LinuxCommandExecutor,
};
const USAGE: &str =
"usage: guanghu-execution-bridge <validate|execute> <request.json> <policy.json>";
fn run() -> Result<(), String> {
let mut arguments = env::args().skip(1);
let mode = arguments.next().ok_or_else(|| USAGE.to_owned())?;
let request_path = arguments.next().ok_or_else(|| USAGE.to_owned())?;
let policy_path = arguments.next().ok_or_else(|| USAGE.to_owned())?;
if arguments.next().is_some() || (mode != "validate" && mode != "execute") {
return Err(USAGE.to_owned());
}
let request: ExecutionRequest = serde_json::from_str(
&fs::read_to_string(&request_path)
.map_err(|error| format!("cannot read request {request_path}: {error}"))?,
)
.map_err(|error| format!("invalid request {request_path}: {error}"))?;
let policy: ExecutionPolicy = serde_json::from_str(
&fs::read_to_string(&policy_path)
.map_err(|error| format!("cannot read policy {policy_path}: {error}"))?,
)
.map_err(|error| format!("invalid policy {policy_path}: {error}"))?;
let plan = compile_request(&request, &policy).map_err(|error| error.to_string())?;
if mode == "validate" {
println!(
"{}",
serde_json::to_string_pretty(&plan)
.map_err(|error| format!("cannot serialize plan: {error}"))?
);
return Ok(());
}
let receipt = execute_plan(&plan, &LinuxCommandExecutor).map_err(|error| error.to_string())?;
println!(
"{}",
serde_json::to_string_pretty(&receipt)
.map_err(|error| format!("cannot serialize receipt: {error}"))?
);
if receipt.final_state == "PASS_100" {
Ok(())
} else {
Err("target-side verification failed".to_owned())
}
}
fn main() -> ExitCode {
match run() {
Ok(()) => ExitCode::SUCCESS,
Err(error) => {
eprintln!("{error}");
ExitCode::FAILURE
}
}
}

View file

@ -0,0 +1,198 @@
use std::{cell::RefCell, collections::VecDeque};
use guanghu_execution_bridge::{
compile_request, execute_plan, ActionKind, AuthorizationReference, CommandExecutor,
CommandResult, ExecutionAction, ExecutionPolicy, ExecutionRequest, RollbackReference,
POLICY_SCHEMA, REQUEST_SCHEMA, REQUIRED_PROTOCOL_CHAIN,
};
fn request(kind: ActionKind) -> ExecutionRequest {
ExecutionRequest {
schema: REQUEST_SCHEMA.to_owned(),
request_id: "REQ-001".to_owned(),
subject_id: "ICE-P-ZY001".to_owned(),
target_node_id: "JD-FD-PRIMARY".to_owned(),
protocol_chain: REQUIRED_PROTOCOL_CHAIN
.iter()
.map(|protocol| (*protocol).to_owned())
.collect(),
action: ExecutionAction {
kind,
resource: "guanghu-broadcast-tower.service".to_owned(),
},
authorization: None,
rollback: None,
}
}
fn policy() -> ExecutionPolicy {
ExecutionPolicy {
schema: POLICY_SCHEMA.to_owned(),
policy_id: "JD-GH-EXEC-001".to_owned(),
target_node_id: "JD-FD-PRIMARY".to_owned(),
allowed_services: vec!["guanghu-broadcast-tower.service".to_owned()],
allow_status: true,
allow_restart: true,
}
}
#[test]
fn compiles_read_only_status_to_exact_systemd_arguments() {
let plan = compile_request(&request(ActionKind::ServiceStatus), &policy()).expect("compile");
assert_eq!(plan.program, "/usr/bin/systemctl");
assert_eq!(
plan.arguments,
["is-active", "guanghu-broadcast-tower.service"]
);
assert!(!plan.mutating);
}
#[test]
fn rejects_missing_protocol_before_execution() {
let mut request = request(ActionKind::ServiceStatus);
request
.protocol_chain
.retain(|protocol| protocol != "GLS-0302");
let error = compile_request(&request, &policy()).expect_err("missing identity must fail");
assert!(error.to_string().contains("GLS-0302"));
}
#[test]
fn rejects_non_allowlisted_service() {
let mut request = request(ActionKind::ServiceStatus);
request.action.resource = "ssh.service".to_owned();
let error = compile_request(&request, &policy()).expect_err("service must fail closed");
assert!(error.to_string().contains("not allowlisted"));
}
#[test]
fn restart_requires_matching_authorization_and_rollback() {
let mut request = request(ActionKind::ServiceRestart);
let error = compile_request(&request, &policy()).expect_err("authorization is required");
assert!(error.to_string().contains("authorization"));
request.authorization = Some(AuthorizationReference {
authorization_id: "AUTH-001".to_owned(),
allowed_action: ActionKind::ServiceRestart,
target_node_id: "JD-FD-PRIMARY".to_owned(),
});
let error = compile_request(&request, &policy()).expect_err("rollback is required");
assert!(error.to_string().contains("rollback"));
request.rollback = Some(RollbackReference {
checkpoint_id: "CHECKPOINT-001".to_owned(),
recovery_action: "restore previous immutable release".to_owned(),
});
let plan = compile_request(&request, &policy()).expect("complete mutation compiles");
assert!(plan.mutating);
assert_eq!(plan.authorization_id.as_deref(), Some("AUTH-001"));
assert_eq!(
plan.rollback_checkpoint_id.as_deref(),
Some("CHECKPOINT-001")
);
}
struct FakeExecutor {
results: RefCell<VecDeque<CommandResult>>,
calls: RefCell<Vec<Vec<String>>>,
}
impl FakeExecutor {
fn new(results: Vec<CommandResult>) -> Self {
Self {
results: RefCell::new(results.into()),
calls: RefCell::new(Vec::new()),
}
}
}
impl CommandExecutor for FakeExecutor {
fn execute(&self, program: &str, arguments: &[String]) -> Result<CommandResult, String> {
let mut call = vec![program.to_owned()];
call.extend(arguments.iter().cloned());
self.calls.borrow_mut().push(call);
self.results
.borrow_mut()
.pop_front()
.ok_or_else(|| "unexpected execution".to_owned())
}
}
#[test]
fn restart_receipt_requires_target_side_active_readback() {
let mut request = request(ActionKind::ServiceRestart);
request.authorization = Some(AuthorizationReference {
authorization_id: "AUTH-001".to_owned(),
allowed_action: ActionKind::ServiceRestart,
target_node_id: "JD-FD-PRIMARY".to_owned(),
});
request.rollback = Some(RollbackReference {
checkpoint_id: "CHECKPOINT-001".to_owned(),
recovery_action: "restore previous immutable release".to_owned(),
});
let plan = compile_request(&request, &policy()).expect("compile restart");
let executor = FakeExecutor::new(vec![
CommandResult {
exit_code: Some(0),
stdout: String::new(),
stderr: String::new(),
},
CommandResult {
exit_code: Some(0),
stdout: "active".to_owned(),
stderr: String::new(),
},
]);
let receipt = execute_plan(&plan, &executor).expect("execute plan");
assert_eq!(receipt.final_state, "PASS_100");
assert!(receipt.target_state_verified);
assert_eq!(executor.calls.borrow().len(), 2);
assert_eq!(executor.calls.borrow()[1][1], "is-active");
}
#[test]
fn successful_restart_command_without_active_readback_is_fail_zero() {
let mut request = request(ActionKind::ServiceRestart);
request.authorization = Some(AuthorizationReference {
authorization_id: "AUTH-001".to_owned(),
allowed_action: ActionKind::ServiceRestart,
target_node_id: "JD-FD-PRIMARY".to_owned(),
});
request.rollback = Some(RollbackReference {
checkpoint_id: "CHECKPOINT-001".to_owned(),
recovery_action: "restore previous immutable release".to_owned(),
});
let plan = compile_request(&request, &policy()).expect("compile restart");
let executor = FakeExecutor::new(vec![
CommandResult {
exit_code: Some(0),
stdout: String::new(),
stderr: String::new(),
},
CommandResult {
exit_code: Some(3),
stdout: "inactive".to_owned(),
stderr: String::new(),
},
]);
let receipt = execute_plan(&plan, &executor).expect("execute plan");
assert_eq!(receipt.final_state, "FAIL_0");
assert!(!receipt.target_state_verified);
}
#[test]
fn rejects_a_deserialized_plan_that_changes_status_into_restart() {
let mut plan =
compile_request(&request(ActionKind::ServiceStatus), &policy()).expect("compile status");
plan.arguments[0] = "restart".to_owned();
let executor = FakeExecutor::new(Vec::new());
let error = execute_plan(&plan, &executor).expect_err("tampered plan must fail");
assert!(error.to_string().contains("not an allowlisted"));
assert!(executor.calls.borrow().is_empty());
}