fix(privacy): avoid repeated full-request processing (#420)

Stage scoped body fragments and reuse context locations while preserving per-rule rollback. Account for encoded size incrementally and use bounded work plus cancellation instead of a fixed wall-clock cutoff.

Co-authored-by: Zyy <zyycn@users.noreply.github.com>
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-13 17:09:33 +08:00
committed by GitHub
co-authored by Zyy Codex
parent b2dab66a80
commit 5b80340135
6 changed files with 432 additions and 77 deletions
@@ -4,15 +4,17 @@ use gateway_core::settings::privacy::PrivacyScope;
use reqwest::header::{HeaderMap, HeaderValue};
use serde_json::Value;
use super::{Budget, rule::CompiledRule};
use super::{Budget, changes::BodyChanges, rule::CompiledRule};
const METADATA_KEYS: &[&str] = &["turnMetadata", "turn_metadata", "x-codex-turn-metadata"];
pub(super) fn apply(
pub(super) fn prepare(
rule: &CompiledRule,
body: &mut Value,
body: &Value,
headers: &mut HeaderMap,
turn_metadata: &mut Option<String>,
changes: &mut BodyChanges,
contexts: &mut ContextIndex,
budget: &mut Budget<'_>,
) -> Result<usize, &'static str> {
match rule.rule.scope {
@@ -30,16 +32,15 @@ pub(super) fn apply(
*value = HeaderValue::from_str(&raw).map_err(|_| "metadata 请求头编码无效")?;
}
}
count += metadata_fields(rule, body, budget)?;
if let Some(client) = body.get_mut("client_metadata") {
count += metadata_fields(rule, client, budget)?;
}
count += metadata_fields(rule, body, changes, budget)?;
Ok(count)
}
PrivacyScope::DesktopGitContext | PrivacyScope::EnvironmentText => {
contexts(rule, body, budget)
contexts.apply(rule, body, changes, budget)
}
PrivacyScope::RequestBody => {
changes.stage(body, "", budget, |next, budget| rule.apply(next, budget))
}
PrivacyScope::RequestBody => rule.apply(body, budget),
PrivacyScope::RequestHeader => {
let name = reqwest::header::HeaderName::from_bytes(rule.rule.selector.as_bytes())
.map_err(|_| "请求头名称无效")?;
@@ -83,17 +84,22 @@ pub(super) fn apply(
fn metadata_fields(
rule: &CompiledRule,
object: &mut Value,
body: &Value,
changes: &mut BodyChanges,
budget: &mut Budget<'_>,
) -> Result<usize, &'static str> {
let mut count = 0;
for key in METADATA_KEYS {
if let Some(value) = object.get_mut(*key) {
match value {
Value::String(raw) => count += metadata(rule, raw, budget)?,
Value::Object(_) => count += rule.apply(value, budget)?,
_ => return Err("metadata 必须是 JSON 对象或编码后的字符串"),
for prefix in ["", "/client_metadata"] {
for key in METADATA_KEYS {
let path = format!("{prefix}/{key}");
if body.pointer(&path).is_none() {
continue;
}
count += changes.stage(body, &path, budget, |next, budget| match next {
Value::String(raw) => metadata(rule, raw, budget),
Value::Object(_) => rule.apply(next, budget),
_ => Err("metadata 必须是 JSON 对象或编码后的字符串"),
})?;
}
}
Ok(count)
@@ -121,74 +127,130 @@ fn metadata(
Ok(count)
}
fn contexts(
rule: &CompiledRule,
body: &mut Value,
budget: &mut Budget<'_>,
) -> Result<usize, &'static str> {
let Some(items) = body.get_mut("input").and_then(Value::as_array_mut) else {
return Ok(0);
};
let mut count = 0;
for item in items {
budget.step()?;
let role = item
.get("role")
.and_then(Value::as_str)
.unwrap_or("")
.to_owned();
#[derive(Default)]
pub(super) struct ContextIndex {
desktop: Option<Vec<ContextTarget>>,
environment: Option<Vec<ContextTarget>>,
}
struct ContextTarget {
path: String,
open: &'static str,
close: &'static str,
identified: bool,
}
impl ContextIndex {
fn apply(
&mut self,
rule: &CompiledRule,
body: &Value,
changes: &mut BodyChanges,
budget: &mut Budget<'_>,
) -> Result<usize, &'static str> {
let desktop = rule.rule.scope == PrivacyScope::DesktopGitContext;
let targets = if desktop {
&mut self.desktop
} else {
&mut self.environment
};
if targets.is_none() {
*targets = Some(locate_contexts(body, desktop, budget)?);
}
let mut count = 0;
for target in targets.iter().flatten() {
budget.step()?;
// 先检查长度再复制,普通对话和工具输出不会进入事务副本
let text = body
.pointer(&target.path)
.and_then(Value::as_str)
.ok_or("上下文文本不存在")?;
let trimmed = text.trim();
let complete = trimmed.starts_with(target.open) && trimmed.ends_with(target.close);
if !complete {
if target.identified {
return Err("已识别上下文不完整,可能已被截断");
}
continue;
}
budget.text(text)?;
count += changes.stage(body, &target.path, budget, |next, budget| {
let Value::String(text) = next else {
return Err("上下文必须是文本");
};
if desktop {
git(rule, text, budget)
} else {
environment(rule, text, budget)
}
})?;
}
Ok(count)
}
}
fn locate_contexts(
body: &Value,
desktop: bool,
budget: &mut Budget<'_>,
) -> Result<Vec<ContextTarget>, &'static str> {
let mut targets = Vec::new();
let Some(items) = body.get("input").and_then(Value::as_array) else {
return Ok(targets);
};
for (item_index, item) in items.iter().enumerate() {
budget.step()?;
let role = item.get("role").and_then(Value::as_str).unwrap_or("");
if !(role == "user" || (!desktop && role == "developer")) {
continue;
}
let (kind, tag) = if desktop {
let (kind, open, close) = if desktop {
(
"additional_content.codex_git_state",
"external_codex_git_state",
"<external_codex_git_state>",
"</external_codex_git_state>",
)
} else if role == "developer" {
(
"additional_content.codex_apps_client_time_context",
"codex_apps_client_time_context",
"<codex_apps_client_time_context>",
"</codex_apps_client_time_context>",
)
} else {
("environments.environment_context", "environment_context")
(
"environments.environment_context",
"<environment_context>",
"</environment_context>",
)
};
let kinds = item
.pointer("/internal_chat_message_metadata_passthrough/content_item_kinds")
.cloned();
let Some(parts) = item.get_mut("content").and_then(Value::as_array_mut) else {
let kinds = item.pointer("/internal_chat_message_metadata_passthrough/content_item_kinds");
let Some(parts) = item.get("content").and_then(Value::as_array) else {
continue;
};
for (index, part) in parts.iter_mut().enumerate() {
if let Some(kinds) = &kinds
&& kinds.get(index).and_then(Value::as_str) != Some(kind)
{
for (index, part) in parts.iter().enumerate() {
budget.step()?;
if kinds.is_some_and(|kinds| kinds.get(index).and_then(Value::as_str) != Some(kind)) {
continue;
}
if part.get("type").and_then(Value::as_str) != Some("input_text") {
continue;
}
let Some(Value::String(text)) = part.get_mut("text") else {
let Some(text) = part.get("text").and_then(Value::as_str) else {
continue;
};
let full = text.trim().starts_with(&format!("<{tag}>"))
&& text.trim().ends_with(&format!("</{tag}>"));
if !full && kinds.is_none() {
let trimmed = text.trim();
if kinds.is_none() && !(trimmed.starts_with(open) && trimmed.ends_with(close)) {
continue;
}
if !full {
return Err("已识别上下文不完整,可能已被截断");
}
budget.text(text)?;
count += if desktop {
git(rule, text, budget)?
} else {
environment(rule, text, budget)?
};
targets.push(ContextTarget {
path: format!("/input/{item_index}/content/{index}/text"),
open,
close,
identified: kinds.is_some(),
});
}
}
Ok(count)
Ok(targets)
}
fn git(
@@ -0,0 +1,69 @@
//! 暂存单条规则改动的正文片段,校验最终大小后统一提交
use serde_json::Value;
use super::{Budget, MAX_REQUEST_BYTES, encoded_size};
pub(super) struct BodyChanges {
original_size: usize,
replacements: Vec<(String, Value)>,
removed_bytes: usize,
added_bytes: usize,
}
impl BodyChanges {
pub(super) fn new(original_size: usize) -> Self {
Self {
original_size,
replacements: Vec::new(),
removed_bytes: 0,
added_bytes: 0,
}
}
pub(super) fn stage(
&mut self,
body: &Value,
path: &str,
budget: &mut Budget<'_>,
apply: impl FnOnce(&mut Value, &mut Budget<'_>) -> Result<usize, &'static str>,
) -> Result<usize, &'static str> {
let original = body.pointer(path).ok_or("目标正文片段不存在")?;
let mut next = original.clone();
let count = apply(&mut next, budget)?;
if count > 0 {
self.removed_bytes += if path.is_empty() {
self.original_size
} else {
encoded_size(original, MAX_REQUEST_BYTES)?
};
// 片段互不重叠,其编码字节总量是最终正文大小的下界
// 提前停止必然超限的改写,避免暂存所有膨胀后的上下文
self.added_bytes = self
.added_bytes
.checked_add(encoded_size(&next, MAX_REQUEST_BYTES)?)
.filter(|size| *size <= MAX_REQUEST_BYTES)
.ok_or("请求或预览样本超出大小限制")?;
self.replacements.push((path.to_owned(), next));
}
Ok(count)
}
pub(super) fn size(&self) -> Result<usize, &'static str> {
self.original_size
.checked_sub(self.removed_bytes)
.and_then(|size| size.checked_add(self.added_bytes))
.filter(|size| *size <= MAX_REQUEST_BYTES)
.ok_or("请求或预览样本超出大小限制")
}
pub(super) fn commit(self, body: &mut Value) {
// 路径来自本条规则执行前的正文,片段互不包含且提交前正文尚未改写
// 仅替换值,不增删这些路径的祖先,因此先前替换不会使后续路径失效
for (path, next) in self.replacements {
if let Some(target) = body.pointer_mut(&path) {
*target = next;
}
}
}
}
@@ -1,18 +1,15 @@
//! OpenAI 隐私规则的统一编译与执行,供设置预览和实际出站共享
mod carrier;
mod changes;
mod rule;
mod selector;
use std::{
collections::BTreeSet,
sync::Arc,
time::{Duration, Instant},
};
use std::{collections::BTreeSet, sync::Arc};
use gateway_core::settings::privacy::{
CodexPrivacyPolicy, CompiledPrivacyPolicy, PrivacyError, PrivacyFailureMode,
PrivacyPreviewRequest, PrivacyPreviewResult, PrivacyRuleOutcome,
PrivacyPreviewRequest, PrivacyPreviewResult, PrivacyRuleOutcome, PrivacyScope,
};
use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
use serde_json::Value;
@@ -36,7 +33,6 @@ impl std::fmt::Debug for Policy {
}
struct Budget<'a> {
started: Instant,
steps: usize,
text_bytes: usize,
cancelled: &'a dyn Fn() -> bool,
@@ -48,8 +44,8 @@ impl Budget<'_> {
if (self.cancelled)() {
return Err("请求已取消或超时");
}
if self.steps > 10_000 || self.started.elapsed() > Duration::from_millis(100) {
return Err("规则执行预算超限");
if self.steps > 10_000 {
return Err("规则执行步数超限");
}
Ok(())
}
@@ -105,11 +101,12 @@ impl CompiledPrivacyPolicy for Policy {
) -> Result<Vec<PrivacyRuleOutcome>, PrivacyError> {
let mut outcomes = Vec::new();
let mut budget = Budget {
started: Instant::now(),
steps: 0,
text_bytes: 0,
cancelled,
};
let mut body_size = None;
let mut contexts = carrier::ContextIndex::default();
for (rule_index, rule) in self.rules.iter().enumerate() {
let mut outcome = PrivacyRuleOutcome {
rule_id: rule.rule.id.clone(),
@@ -121,22 +118,30 @@ impl CompiledPrivacyPolicy for Policy {
// 一条规则的所有承载副本同成同败,失败不会留下部分改写
let result = (|| {
budget.step()?;
check_size(body, MAX_REQUEST_BYTES)?;
let mut next_body = body.clone();
let original_size =
*body_size.get_or_insert_with(|| encoded_size(body, MAX_REQUEST_BYTES));
let original_size = original_size?;
let mut next_headers = headers.clone();
let mut next_metadata = turn_metadata.clone();
let count = carrier::apply(
let mut changes = changes::BodyChanges::new(original_size);
let count = carrier::prepare(
rule,
&mut next_body,
body,
&mut next_headers,
&mut next_metadata,
&mut changes,
&mut contexts,
&mut budget,
)?;
check_size(&next_body, MAX_REQUEST_BYTES)?;
let next_size = changes.size()?;
budget.step()?;
*body = next_body;
changes.commit(body);
*headers = next_headers;
*turn_metadata = next_metadata;
body_size = Some(Ok(next_size));
if count > 0 && rule.rule.scope == PrivacyScope::RequestBody {
contexts = carrier::ContextIndex::default();
}
Ok(count)
})();
match result {
@@ -223,6 +228,10 @@ pub fn preview(request: PrivacyPreviewRequest) -> Result<PrivacyPreviewResult, P
}
fn check_size(value: &Value, limit: usize) -> Result<(), &'static str> {
encoded_size(value, limit).map(|_| ())
}
fn encoded_size(value: &Value, limit: usize) -> Result<usize, &'static str> {
struct Counter(usize);
impl std::io::Write for Counter {
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
@@ -236,7 +245,9 @@ fn check_size(value: &Value, limit: usize) -> Result<(), &'static str> {
Ok(())
}
}
serde_json::to_writer(Counter(limit), value).map_err(|_| "请求或预览样本超出大小限制")
let mut counter = Counter(limit);
serde_json::to_writer(&mut counter, value).map_err(|_| "请求或预览样本超出大小限制")?;
Ok(limit - counter.0)
}
impl super::client::CodexBackendClient {
@@ -0,0 +1,210 @@
//! 局部隐私事务的跨副本回滚、上下文定位失效与编码大小边界
use super::{policy, rule, sample};
use gateway_core::settings::privacy::*;
use provider_openai::transport::privacy::{compile, preview};
use serde_json::{Value, json};
fn git_item(value: Value) -> Value {
json!({"role":"user","content":[{"type":"input_text","text":format!(
"<external_codex_git_state><environment_context><git>{value}</git></environment_context></external_codex_git_state>"
)}]})
}
#[test]
fn failed_metadata_rule_preserves_body_headers_and_independent_metadata() {
let mut request = sample(
vec![rule(
PrivacyScope::TurnMetadata,
"$.secret",
PrivacyAction::RemoveField,
)],
json!({"turnMetadata":{"secret":true},"client_metadata":{"turnMetadata":42}}),
);
request.policy.on_error = PrivacyFailureMode::SkipRule;
request.turn_metadata = Some(r#"{"secret":true}"#.into());
request.headers.insert(
"x-codex-turn-metadata".into(),
vec![r#"{"secret":true}"#.into()],
);
let output = preview(request.clone()).unwrap();
assert_eq!(output.body, request.body);
assert_eq!(output.headers, request.headers);
assert_eq!(output.turn_metadata, request.turn_metadata);
assert_eq!(output.outcomes[0].status, "skipped");
}
#[test]
fn malformed_later_git_copy_rolls_back_every_context() {
let first = git_item(json!({"secret":true}));
let mut second = first.clone();
second["content"][0]["text"] = json!(
"<external_codex_git_state><environment_context><git>invalid</git></environment_context></external_codex_git_state>"
);
let body = json!({"input":[first,second]});
let mut request = sample(
vec![rule(
PrivacyScope::DesktopGitContext,
"$.secret",
PrivacyAction::RemoveField,
)],
body.clone(),
);
request.policy.on_error = PrivacyFailureMode::SkipRule;
let output = preview(request).unwrap();
assert_eq!(output.body, body);
assert_eq!(output.outcomes[0].status, "skipped");
}
#[test]
fn body_deletion_invalidates_context_positions_between_rules() {
let first = rule(
PrivacyScope::DesktopGitContext,
"$.first",
PrivacyAction::RemoveField,
);
let mut delete = rule(
PrivacyScope::RequestBody,
"$.input[0]",
PrivacyAction::RemoveField,
);
delete.id = "delete".into();
let mut last = rule(
PrivacyScope::DesktopGitContext,
"$.last",
PrivacyAction::RemoveField,
);
last.id = "last".into();
let output = preview(sample(
vec![first, delete, last],
json!({"input":[
git_item(json!({"first":1})),git_item(json!({"first":1,"last":2,"keep":true}))
]}),
))
.unwrap();
assert_eq!(
output
.outcomes
.iter()
.map(|o| o.matches)
.collect::<Vec<_>>(),
vec![2, 1, 1]
);
assert_eq!(
output.body,
json!({"input":[git_item(json!({"keep":true}))]})
);
}
#[test]
fn cached_candidates_recheck_context_wrapper_after_text_replacement() {
let mut first = rule(
PrivacyScope::EnvironmentText,
"$",
PrivacyAction::RegexReplace,
);
first.pattern = Some("<environment_context>".into());
first.replacement = "<environment_context note=\"test\">".into();
let mut last = rule(
PrivacyScope::EnvironmentText,
"$",
PrivacyAction::RegexReplace,
);
last.id = "last".into();
last.pattern = Some("private".into());
last.replacement = "public".into();
let output = preview(sample(
vec![first, last],
json!({"input":[{"role":"user","content":[
{"type":"input_text","text":"<environment_context>private</environment_context>"}
]}]}),
))
.unwrap();
assert_eq!(output.outcomes[0].status, "applied");
assert_eq!(output.outcomes[1].status, "unmatched");
assert!(
output.body["input"][0]["content"][0]["text"]
.as_str()
.unwrap()
.contains("private")
);
}
#[test]
fn unicode_reencoding_growth_rejects_and_rolls_back_only_current_rule() {
let mut remove = rule(
PrivacyScope::TurnMetadata,
"$.secret",
PrivacyAction::RemoveField,
);
remove.id = "remove".into();
let mut last = rule(
PrivacyScope::RequestHeader,
"x-remove",
PrivacyAction::RemoveField,
);
last.id = "last".into();
let mut config = policy(vec![remove, last]);
config.on_error = PrivacyFailureMode::SkipRule;
let compiled = compile(&config).unwrap();
let mut body = json!({"turnMetadata":json!({"secret":true,"keep":"界".repeat(100)}).to_string(),"padding":""});
let padding = 32 * 1024 * 1024 - serde_json::to_vec(&body).unwrap().len();
body["padding"] = json!("a".repeat(padding));
let original = body.clone();
let mut headers = reqwest::header::HeaderMap::new();
headers.insert("x-remove", "test".parse().unwrap());
let outcomes = compiled
.apply(&mut body, &mut headers, &mut None, &|| false)
.unwrap();
assert_eq!(body, original);
assert_eq!(
outcomes[0].reason.as_deref(),
Some("请求或预览样本超出大小限制")
);
assert_eq!(outcomes[1].status, "applied");
assert!(headers.is_empty());
}
#[test]
fn cancellation_after_staging_leaves_every_carrier_unchanged() {
let config = policy(vec![rule(
PrivacyScope::TurnMetadata,
"$.secret",
PrivacyAction::RemoveField,
)]);
let compiled = compile(&config).unwrap();
let original =
json!({"turnMetadata":{"secret":true},"client_metadata":{"turnMetadata":{"secret":true}}});
// 覆盖每一个取消检查点,包括片段已暂存、尚未提交的最后一步
let mut successful = false;
for cancel_at in 1..64 {
let mut body = original.clone();
let mut headers = reqwest::header::HeaderMap::new();
headers.insert(
"x-codex-turn-metadata",
r#"{"secret":true}"#.parse().unwrap(),
);
let original_headers = headers.clone();
let mut metadata = Some(r#"{"secret":true}"#.to_owned());
let original_metadata = metadata.clone();
let calls = std::cell::Cell::new(0);
let result = compiled.apply(&mut body, &mut headers, &mut metadata, &|| {
calls.set(calls.get() + 1);
calls.get() >= cancel_at
});
match result {
Err(error) => {
assert_eq!(error.reason, "请求已取消或超时");
assert_eq!(body, original);
assert_eq!(headers, original_headers);
assert_eq!(metadata, original_metadata);
}
Ok(outcomes) => {
assert_eq!(outcomes[0].matches, 4);
successful = true;
break;
}
}
}
assert!(successful);
}
@@ -1,5 +1,7 @@
//! 隐私规则的多承载一致性、自由字段配置、条件删除与有界失败合同
mod changes;
use gateway_core::settings::privacy::*;
use provider_openai::transport::privacy::{compile, preview};
use serde_json::{Value, json};
@@ -291,7 +293,7 @@ fn privacy_validates_regex_paths_captures_and_execution_limits() {
let error = policy
.apply(&mut body, &mut Default::default(), &mut None, &|| false)
.unwrap_err();
assert_eq!(error.reason, "规则执行预算超限");
assert_eq!(error.reason, "规则执行步数超限");
assert_eq!(body, original);
assert!(
policy
+2 -1
View File
@@ -1433,7 +1433,8 @@ JSON 路径支持 `$`、`.成员`、`["带特殊字符的键"]`、`[索引]`、`
`replaceAll` 控制首个或全部匹配,另支持忽略大小写和多行锚点。最多 32 条规则,路径最多 16 层、1024 字节,
模式与替换内容分别最多 4096 字节,固定值编码后最多 4096 字节。XML 上下文最多 10000 个节点,环境文本最多 64 层。
执行限制为正文 32 MiB、单目标文本及编码后的上下文 1 MiB、
累计处理文本 16 MiB、10000 次访问与匹配步骤、100 ms 检查预算;时间预算在处理检查点检查,不是硬实时中断
累计处理文本 16 MiB、每次执行 10000 次访问与匹配步骤。
执行检查点响应请求取消,步数超限报告「规则执行步数超限」,不以固定墙钟时长判定规则失败
执行失败时,`skip_rule` 撤销本条规则在所有副本上的修改并继续,意味着本条脱敏未完成;
`reject_request` 阻止发送并返回本地请求错误,不因该错误自动重试或换号。设置保存时校验所有规则,包含未启用的规则