feat: 完成 OpenNexus 第三阶段核心功能与生产化基础 #45

Merged
Kronecker merged 151 commits from feat/phase3-completion into main 2026-09-10 01:16:31 +08:00
4 changed files with 542 additions and 0 deletions
Showing only changes of commit 0a347bee6d - Show all commits
@@ -447,3 +447,15 @@ Core 的独立数据目录目前不等于已授权 Vault。Python 旧笔记写
- 测试验证无秘密包在初始锁定时拒绝,解锁后能准备,准备后锁定则不能创建进程;锁定期间新签发的许可也不能用于准备。真实 AppContainer 无秘密双进程实例在锁定后约 55.14 ms 清空,后续工具调用被拒绝。
- 53 项扩展回归通过,4 项 ignored 为三个父测试实际驱动的 Job 辅助入口及单独执行的 60 秒验收;全目标 Clippy -D warnings 通过。日志 `.build/extension-locked-session-tests.log``.build/extension-locked-session-clippy.log`
- 这修复会话门禁缺口,尚未开放执行能力或完成实际桌面实例注册、broker、CPU/scratch 限额及整体验收;完整生产化目标继续进行。
## 增量:实例文件 broker 策略及写入意图提交前复核
- 新增 Windows 桌面内部 extension_file_broker,绑定已签发 Claims、当前 Vault、平台/策略、权限及已解锁会话租约。只接受 notes.read / notes.write 严格 DTO,不接受请求自报 Vault、来源、权限或 Core 任意方法;传输层仍须在创建 broker 前完成真实实例身份认证。
- 每实例每秒最多 32 次请求,畸形或过大请求也占用额度;JSON 帧上限 2 MiB、笔记正文上限 1 MiB、路径上限 1024 字节。权限在文件访问前检查,只允许合法 Markdown 路径并沿用 Vault 私有路径禁令。原生传输尚未接入,未来读取帧时必须先执行大小限制,不能读完无界内容后才调用此适配器。
- 读取从持有的目录句柄逐级 nofollow 打开,文件只共享读取;核对普通文件、非重解析对象及单硬链接,并以有界读取再次约束长度。返回实际正文摘要用于 expected_hash;返回前复核租约。
- 写入通过 Host CAS / 持久日志 / outbox 完成。操作 UUID 按执行类型、来源、命名空间、包 ID、Vault 派生隔离标识,避免不同包或 Core/renderer 直接复用同一操作 ID;响应保留调用方原 ID。同包同 Vault 重启 broker 与 Workspace 后可重放已提交回执,变更同 ID 内容返回冲突。
- Workspace 增加内部 guarded 写入入口,进入处理及持久写入意图事务提交前检查授权。后一次拒绝会回滚 operations / journal,重启不能重放该写入;此前已刷盘的 payload 及其映射可能留下,需后续统一孤儿清理。已接受的持久意图继续沿用恢复语义,不在写入文件后因撤销跳过 outbox。
- 新增真实临时 Workspace / Stronghold 测试,覆盖正常读写、CAS、同 ID 内容冲突、跨包隔离、重启重放、权限/伪造字段/私有路径拒绝、帧/正文/速率限制、硬链接与异 Vault 拒绝、凭据锁定撤销。另有 Workspace 故障注入测试在提交前拒绝,确认重启正文与 outbox 不变,同 ID 原内容可以重试并只增加一次 revision/outbox。
- 当前仍是内部策略适配器,未开放第三方执行。写入底层尚使用字符串路径,需完成全程句柄约束;原子撤权与最终提交的严格排序、真实管道身份认证、Vault 根目录替换、网络 broker 及完整攻击矩阵也尚未完成,不据此判定 C-02 / C-04 或整体生产化通过。
- 本轮完整 Rust desktop all-targets 回归 115 项通过、6 项 ignored;其中三个 Job 辅助入口和一个上传故障辅助入口由父测试实际驱动,另外两项是需显式执行的 60 秒期限验收与打包 Core 20 次冷启动,本轮未重跑。全目标 Clippy -D warnings 通过。日志 `.build/extension-file-broker-full-tests.log``.build/extension-file-broker-clippy.log`。未修改前端/Python 实现,本轮没有把此前测试结果冒充重新执行结果。
@@ -0,0 +1,436 @@
//! Instance-bound file RPC policy. Transport must bind one Broker to one
//! authenticated instance. Write commits currently use Workspace's transaction;
//! full handle-relative write hardening is required before untrusted activation.
use crate::{
credentials::CredentialBroker,
extension_permit::{Authority, Claims, Lease, Permit},
workspace::{HostError, Result, Workspace},
};
use cap_fs_ext::{DirExt, FollowSymlinks, OpenOptionsFollowExt};
use cap_std::fs::{Dir, OpenOptions, OpenOptionsExt};
use serde::Deserialize;
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use std::{
collections::BTreeSet,
io::Read,
os::windows::io::AsRawHandle,
time::{Duration, Instant},
};
use windows_sys::Win32::Storage::FileSystem::*;
pub const MAX_FRAME_BYTES: usize = 2 * 1024 * 1024;
pub const MAX_NOTE_BYTES: usize = 1024 * 1024;
const REQUESTS_PER_SECOND: u32 = 32;
#[derive(Deserialize)]
#[serde(tag = "method", deny_unknown_fields)]
enum Request {
#[serde(rename = "notes.read")]
Read { path: String },
#[serde(rename = "notes.write")]
Write {
path: String,
expected_hash: String,
content: String,
operation_id: String,
},
}
pub struct Broker {
root: Dir,
vault: String,
permissions: BTreeSet<String>,
lease: Lease,
identity: Vec<u8>,
window: Instant,
requests: u32,
}
impl Broker {
/// Called by the Host after instance identity binding; never an IPC command.
pub fn bind(
authority: &Authority,
permit: &Permit,
claims: &Claims,
credentials: &CredentialBroker,
workspace: &Workspace,
policy_version: &str,
now_ms: u64,
) -> Result<Self> {
let mut lease = authority.lease(permit, claims, now_ms)?;
lease.bind_credential(credentials.lock_signal());
if credentials.is_locked() {
return Err(HostError::new("CREDENTIALS_LOCKED"));
}
if claims.platform != std::env::consts::OS || claims.policy_version != policy_version {
return Err(HostError::new("EXTENSION_EXECUTION_CONTEXT_CHANGED"));
}
if claims.vault_id != workspace.vault_id {
return Err(HostError::new("VAULT_PERMISSION_CHANGED"));
}
let root = Dir::open_ambient_dir(&workspace.root, cap_std::ambient_authority())?;
let identity = serde_json::to_vec(&(
claims.kind,
&claims.source,
&claims.namespace,
&claims.package_id,
&claims.vault_id,
))
.map_err(|_| HostError::new("EXTENSION_BROKER_INVALID"))?;
lease.check()?;
Ok(Self {
root,
vault: claims.vault_id.clone(),
permissions: claims.permissions.clone(),
lease,
identity,
window: Instant::now(),
requests: 0,
})
}
fn operation(&self, operation: &str) -> Result<String> {
let id =
uuid::Uuid::parse_str(operation).map_err(|_| HostError::new("OPERATION_ID_INVALID"))?;
if id.to_string() != operation {
return Err(HostError::new("OPERATION_ID_INVALID"));
}
let mut digest = Sha256::new();
digest.update(b"OpenNexus extension note operation v1\0");
digest.update((self.identity.len() as u64).to_be_bytes());
digest.update(&self.identity);
digest.update(id.as_bytes());
let mut bytes: [u8; 16] = digest.finalize()[..16].try_into().unwrap();
bytes[6] = (bytes[6] & 15) | 128;
bytes[8] = (bytes[8] & 63) | 128;
Ok(uuid::Uuid::from_bytes(bytes).to_string())
}
fn permit(&self, permission: &str) -> Result<()> {
if !self.permissions.contains(permission) {
return Err(HostError::new("EXTENSION_PERMISSION_DENIED"));
}
self.lease.check()
}
/// The transport must apply MAX_FRAME_BYTES while reading, before allocation.
/// Hold the Host workspace lock throughout dispatch and commit.
pub fn dispatch(&mut self, workspace: &mut Workspace, bytes: &[u8]) -> Result<Value> {
self.lease.check()?;
if workspace.vault_id != self.vault {
return Err(HostError::new("VAULT_PERMISSION_CHANGED"));
}
if self.window.elapsed() >= Duration::from_secs(1) {
self.window = Instant::now();
self.requests = 0;
}
if self.requests >= REQUESTS_PER_SECOND {
return Err(HostError::new("EXTENSION_BROKER_RATE_LIMITED"));
}
self.requests += 1;
if bytes.len() > MAX_FRAME_BYTES {
return Err(HostError::new("EXTENSION_BROKER_REQUEST_TOO_LARGE"));
}
let request: Request = serde_json::from_slice(bytes)
.map_err(|_| HostError::new("EXTENSION_BROKER_REQUEST_INVALID"))?;
self.permit(match &request {
Request::Read { .. } => "notes.read",
Request::Write { .. } => "notes.write",
})?;
let path = match &request {
Request::Read { path } | Request::Write { path, .. } => path,
};
if path.len() > 1024
|| !path.to_ascii_lowercase().ends_with(".md")
|| path
.split('/')
.any(|part| part.is_empty() || part == "." || part == "..")
{
return Err(HostError::new("EXTENSION_BROKER_PATH_INVALID"));
}
workspace.resolve(path)?;
match request {
Request::Read { path } => {
let mut parent = self.root.try_clone()?;
let mut parts = path.split('/').peekable();
let mut leaf = None;
while let Some(part) = parts.next() {
if part.is_empty() || part == "." || part == ".." {
return Err(HostError::new("UNSAFE_PATH"));
}
if parts.peek().is_none() {
leaf = Some(part);
} else {
parent = parent.open_dir_nofollow(part)?;
}
}
let mut options = OpenOptions::new();
options
.read(true)
.follow(FollowSymlinks::No)
.share_mode(FILE_SHARE_READ);
let file = parent
.open_with(leaf.ok_or_else(|| HostError::new("UNSAFE_PATH"))?, &options)?
.into_std();
let metadata = file.metadata()?;
let mut info = BY_HANDLE_FILE_INFORMATION::default();
if !metadata.is_file()
|| unsafe { GetFileInformationByHandle(file.as_raw_handle(), &mut info) } == 0
|| info.nNumberOfLinks != 1
|| info.dwFileAttributes & FILE_ATTRIBUTE_REPARSE_POINT != 0
{
return Err(HostError::new("UNSAFE_PATH"));
}
if metadata.len() > MAX_NOTE_BYTES as u64 {
return Err(HostError::new("EXTENSION_NOTE_TOO_LARGE"));
}
let mut content = Vec::new();
file.take(MAX_NOTE_BYTES as u64 + 1)
.read_to_end(&mut content)?;
if content.len() > MAX_NOTE_BYTES {
return Err(HostError::new("EXTENSION_NOTE_TOO_LARGE"));
}
let content = String::from_utf8(content)
.map_err(|_| HostError::new("EXTENSION_NOTE_ENCODING_INVALID"))?;
self.lease.check()?;
Ok(
json!({"path":path,"expected_hash":format!("{:x}",Sha256::digest(content.as_bytes())),"content":content}),
)
}
Request::Write {
path,
expected_hash,
content,
operation_id,
} => {
if content.len() > MAX_NOTE_BYTES {
return Err(HostError::new("EXTENSION_NOTE_TOO_LARGE"));
}
if !expected_hash.is_empty()
&& (expected_hash.len() != 64
|| !expected_hash
.bytes()
.all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)))
{
return Err(HostError::new("EXTENSION_BROKER_REQUEST_INVALID"));
}
let operation = self.operation(&operation_id)?;
self.lease.check()?;
let entry = workspace.write_operation_guarded(
&path,
&expected_hash,
content.as_bytes(),
&operation,
|| self.lease.check(),
)?;
Ok(json!({"operation_id":operation_id,"state":"committed","result":entry}))
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::extension_permit::ExecutionKind;
use zeroize::Zeroizing;
fn setup() -> (
tempfile::TempDir,
Workspace,
CredentialBroker,
Authority,
Claims,
) {
let temp = tempfile::tempdir().unwrap();
let vault = temp.path().join("vault");
std::fs::create_dir(&vault).unwrap();
let mut ws = Workspace::open(&vault).unwrap();
ws.write("note.md", "", b"original", "local").unwrap();
let mut credentials = CredentialBroker::new(temp.path().join("credentials.v1"));
credentials
.unlock(Zeroizing::new(b"file broker fixture password".to_vec()))
.unwrap();
let claims = Claims {
kind: ExecutionKind::Mcp,
source: "https://catalog.example/".into(),
namespace: "examples".into(),
package_id: "reviewer".into(),
version: "1.0.0".into(),
archive_sha256: "a".repeat(64),
tree_sha256: "b".repeat(64),
signer_sha256: "c".repeat(64),
entry: "entry.exe".into(),
arguments: vec![],
environment: Default::default(),
permissions: ["notes.read".into(), "notes.write".into()]
.into_iter()
.collect(),
vault_id: ws.vault_id.clone(),
platform: "windows".into(),
policy_version: "1".into(),
expires_at_ms: 120_000,
};
(temp, ws, credentials, Authority::default(), claims)
}
fn call(broker: &mut Broker, ws: &mut Workspace, request: Value) -> Result<Value> {
broker.dispatch(ws, &serde_json::to_vec(&request).unwrap())
}
#[test]
fn bound_read_write_cas_and_scoped_operation_replay_use_host_journal() {
let (_temp, mut ws, credentials, authority, claims) = setup();
let permit = authority.issue(&claims, 1).unwrap();
let mut broker =
Broker::bind(&authority, &permit, &claims, &credentials, &ws, "1", 2).unwrap();
let read = call(
&mut broker,
&mut ws,
json!({"method":"notes.read","path":"note.md"}),
)
.unwrap();
assert_eq!(read["content"], "original");
let operation = uuid::Uuid::new_v4().to_string();
let write = json!({"method":"notes.write","path":"note.md","expected_hash":read["expected_hash"],"content":"extension update","operation_id":operation});
let receipt = call(&mut broker, &mut ws, write.clone()).unwrap();
let revision = receipt["result"]["revision"].clone();
assert_eq!(call(&mut broker, &mut ws, write.clone()).unwrap(), receipt);
assert_eq!(ws.read("note.md").unwrap().content, "extension update");
let mut altered = write.clone();
altered["content"] = json!("changed payload");
assert_eq!(
call(&mut broker, &mut ws, altered).unwrap_err().code,
"OPERATION_PAYLOAD_CONFLICT"
);
// Build a stale CAS with a genuinely new operation ID.
let mut stale = write.clone();
stale["operation_id"] = json!(uuid::Uuid::new_v4().to_string());
assert_eq!(
call(&mut broker, &mut ws, stale).unwrap_err().code,
"REVISION_CONFLICT"
);
let scoped = broker.operation(&operation).unwrap();
assert!(ws.operation(&operation).unwrap().is_none());
assert_eq!(
ws.operation(&scoped).unwrap().unwrap()["result"]["revision"],
revision
);
let mut another = claims.clone();
another.package_id = "another".into();
let other_permit = authority.issue(&another, 1).unwrap();
let other = Broker::bind(
&authority,
&other_permit,
&another,
&credentials,
&ws,
"1",
2,
)
.unwrap();
assert_ne!(other.operation(&operation).unwrap(), scoped);
drop(other);
drop(broker);
let root = ws.root.clone();
drop(ws);
let mut ws = Workspace::open(&root).unwrap();
let mut replay =
Broker::bind(&authority, &permit, &claims, &credentials, &ws, "1", 2).unwrap();
assert_eq!(call(&mut replay, &mut ws, write).unwrap(), receipt);
}
#[test]
fn permissions_scope_limits_hardlinks_and_revocation_fail_closed() {
let (_temp, mut ws, mut credentials, authority, mut claims) = setup();
claims.permissions.clear();
let permit = authority.issue(&claims, 1).unwrap();
let mut denied =
Broker::bind(&authority, &permit, &claims, &credentials, &ws, "1", 2).unwrap();
assert_eq!(
call(
&mut denied,
&mut ws,
json!({"method":"notes.read","path":"note.md"})
)
.unwrap_err()
.code,
"EXTENSION_PERMISSION_DENIED"
);
claims.permissions.insert("notes.read".into());
let permit = authority.issue(&claims, 1).unwrap();
let mut broker =
Broker::bind(&authority, &permit, &claims, &credentials, &ws, "1", 2).unwrap();
assert_eq!(
call(
&mut broker,
&mut ws,
json!({"method":"notes.read","path":"note.md","vault_id":"forged"})
)
.unwrap_err()
.code,
"EXTENSION_BROKER_REQUEST_INVALID"
);
assert_eq!(
call(
&mut broker,
&mut ws,
json!({"method":"credentials.resolve"})
)
.unwrap_err()
.code,
"EXTENSION_BROKER_REQUEST_INVALID"
);
assert!(call(
&mut broker,
&mut ws,
json!({"method":"notes.read","path":".ainote/private.md"})
)
.is_err());
assert_eq!(
broker
.dispatch(&mut ws, &vec![b' '; MAX_FRAME_BYTES + 1])
.unwrap_err()
.code,
"EXTENSION_BROKER_REQUEST_TOO_LARGE"
);
std::fs::write(ws.root.join("large.md"), vec![b'x'; MAX_NOTE_BYTES + 1]).unwrap();
assert_eq!(
call(
&mut broker,
&mut ws,
json!({"method":"notes.read","path":"large.md"})
)
.unwrap_err()
.code,
"EXTENSION_NOTE_TOO_LARGE"
);
let outside = tempfile::tempdir().unwrap();
std::fs::hard_link(ws.root.join("note.md"), outside.path().join("alias.md")).unwrap();
assert_eq!(
call(
&mut broker,
&mut ws,
json!({"method":"notes.read","path":"note.md"})
)
.unwrap_err()
.code,
"UNSAFE_PATH"
);
let second = tempfile::tempdir().unwrap();
let mut other = Workspace::open(second.path()).unwrap();
assert_eq!(
broker.dispatch(&mut other, b"{}").unwrap_err().code,
"VAULT_PERMISSION_CHANGED"
);
let mut limited =
Broker::bind(&authority, &permit, &claims, &credentials, &ws, "1", 2).unwrap();
for _ in 0..REQUESTS_PER_SECOND {
assert_eq!(
limited.dispatch(&mut ws, b"{}").unwrap_err().code,
"EXTENSION_BROKER_REQUEST_INVALID"
);
}
assert_eq!(
limited.dispatch(&mut ws, b"{}").unwrap_err().code,
"EXTENSION_BROKER_RATE_LIMITED"
);
credentials.lock();
assert_eq!(
broker.dispatch(&mut ws, b"{}").unwrap_err().code,
"CREDENTIALS_LOCKED"
);
assert!(Broker::bind(&authority, &permit, &claims, &credentials, &ws, "1", 2).is_err());
}
}
+3
View File
@@ -75,3 +75,6 @@ pub mod extension_launch_authorization;
#[cfg(all(windows, feature = "desktop"))]
mod extension_revocation;
#[cfg(all(windows, feature = "desktop"))]
pub mod extension_file_broker;
+91
View File
@@ -407,6 +407,50 @@ impl Workspace {
operation_id: &str,
identity: Option<&str>,
) -> Result<Entry> {
self.write_authorized(
path,
expected,
content,
origin,
operation_id,
(identity, &|| Ok(())),
)
}
/// Revalidate the caller before work and immediately before committing the
/// durable write intent. Once accepted, recovery must finish that intent.
/// The caller must serialize the authorization boundary if strict atomic
/// ordering with concurrent revocation is required.
#[cfg(any(test, all(windows, feature = "desktop")))]
pub(crate) fn write_operation_guarded(
&mut self,
path: &str,
expected: &str,
content: &[u8],
operation_id: &str,
authorize: impl Fn() -> Result<()>,
) -> Result<Entry> {
self.write_authorized(
path,
expected,
content,
"local",
operation_id,
(None, &authorize),
)
}
fn write_authorized(
&mut self,
path: &str,
expected: &str,
content: &[u8],
origin: &str,
operation_id: &str,
authorization: (Option<&str>, &dyn Fn() -> Result<()>),
) -> Result<Entry> {
let (identity, authorize) = authorization;
authorize()?;
if Uuid::parse_str(operation_id).is_err() {
return Err(HostError::new("OPERATION_ID_INVALID"));
}
@@ -517,6 +561,9 @@ impl Workspace {
origin
],
)?;
// Rejection drops the uncommitted transaction: no recoverable write or
// outbox entry is published. An unreferenced payload is never replayed.
authorize()?;
tx.commit()?;
self.apply_journal(operation_id, &file_id, path, expected, content, origin)?;
self.entry(path)?
@@ -1081,4 +1128,48 @@ mod tests {
.unwrap();
assert_eq!(state, "conflict");
}
#[test]
fn rejected_write_intent_is_not_recovered_and_can_retry() {
use std::cell::Cell;
let dir = tempfile::tempdir().unwrap();
let mut ws = Workspace::open(dir.path()).unwrap();
let initial = ws.write("note.md", "", b"original", "local").unwrap();
let pending = ws.pending_count().unwrap();
let id = Uuid::new_v4().to_string();
let checks = Cell::new(0);
let error = ws
.write_operation_guarded("note.md", &initial.hash, b"update", &id, || {
checks.set(checks.get() + 1);
if checks.get() == 2 {
Err(HostError::new("EXTENSION_PERMIT_REVOKED"))
} else {
Ok(())
}
})
.unwrap_err();
assert_eq!(error.code, "EXTENSION_PERMIT_REVOKED");
assert_eq!(checks.get(), 2);
assert!(ws.operation(&id).unwrap().is_none());
assert_eq!(ws.pending_count().unwrap(), pending);
drop(ws);
let mut ws = Workspace::open(dir.path()).unwrap();
assert_eq!(ws.read("note.md").unwrap().content, "original");
assert_eq!(ws.pending_count().unwrap(), pending);
assert!(ws.operation(&id).unwrap().is_none());
let committed = ws
.write_operation_guarded("note.md", &initial.hash, b"update", &id, || Ok(()))
.unwrap();
assert_eq!(committed.revision, initial.revision + 1);
assert_eq!(ws.pending_count().unwrap(), pending + 1);
// Revoked callers cannot obtain an existing successful receipt either.
assert_eq!(
ws.write_operation_guarded("note.md", &initial.hash, b"update", &id, || Err(
HostError::new("EXTENSION_PERMIT_REVOKED")
))
.unwrap_err()
.code,
"EXTENSION_PERMIT_REVOKED"
);
assert_eq!(ws.read("note.md").unwrap().content, "update");
}
}