helix_driver_host/
http_cross.rs

1//! http_cross — 出站 HTTP 横切层（鉴权注入 / outcome observer / traceparent 透传）。
2//!
3//! ## 归属（P1.5，ADR-007 双端共享）
4//!
5//! 现网（cses-client）这套横切在浏览器 Angular `networkInterceptor` + `app-http.service`：
6//! 鉴权头注入 / 401→logout / 断网提示 / traceparent 出站透传。接管后鉴权与 OTel 执行下沉
7//! host，HTTP outcome 的业务解释由模块注入；PC native 与 Flutter FFI 复用同一装饰器，避免
8//! 两端执行语义漂移。
9//!
10//! ## 设计：装饰器（decorator over `HttpRequester`）
11//!
12//! `CrossCuttingHttp<H, E>` 包内层 `H: HttpRequester`（真出站，如 `SharedHttpClient`）+
13//! `E: EventSink`（业务事件回报通道）+ `AuthTokenRegistry`（driver 持有的真 token）+
14//! `HttpOutcomeObserver`（由业务模块在装配点注入）。它**自身**实现 `HttpRequester`，故能透明
15//! 插进事件泵的 `http` 端口位置，对 core 不可见。
16//!
17//! 每条出站请求：
18//! 1. **鉴权注入**：读 core 标的意图头 `X-Auth-Kind: session|bot`（[`helix_core::AUTH_KIND_HEADER`]），
19//!    据此从 registry 取真 token 注入对应鉴权头，并 **strip 意图头**（绝不出网，HX-C001/C6）。
20//!    core/helix-im 永不持 token——`grep 'mp_|Authorization|Bearer'` 在 core/im 必空。
21//! 2. **traceparent 透传**：core 已把 `traceparent` 放进 `req.headers`（链路头），横切层
22//!    **原样保留**到出站请求——不改不删（出站头 == 入站信封 trace，端到端可断言）。
23//! 3. **结果观察**：把完整 HTTP outcome 交给业务 observer；若返回业务事件则 emit。host 不认识
24//!    事件名或 payload，错误仍原样上抛（不吞）。
25
26use std::sync::Arc;
27
28use parking_lot::RwLock; // 无 poison：panic 不静默丢 token（对齐 network.rs HostHeaderRegistry）
29
30use crate::base64_encode;
31use crate::otel::{HostOtelRuntime, HostSpanScope, TraceDirection};
32use crate::trace::TraceCarrier;
33use helix_core::auth::{apply_auth_intent, AuthKind};
34use helix_core::effect::{DomainEventBytes, HttpRequest, HttpResponse};
35use helix_core::ports::{EventSink, HttpRequester};
36use helix_core::PortError;
37
38/// 平台 HTTP outcome 到业务事件的观察函数。
39///
40/// 规则与 payload 归业务模块；host 只调用它并把返回事件送入 `EventSink`。
41pub type HttpOutcomeObserver =
42    fn(url: &str, outcome: &Result<HttpResponse, PortError>) -> Option<DomainEventBytes>;
43
44/// driver 持有的真鉴权 token（平台壳登录后写入；core 永不触碰）。
45///
46/// 两类凭据各一槽，按 [`AuthKind`] 路由。`Arc<RwLock<_>>` 运行时可热换（token 轮换 / 重登），
47/// 与出站请求并发安全；写少读多，`parking_lot::RwLock` 足够。
48#[derive(Clone, Default)]
49pub struct AuthTokenRegistry {
50    inner: Arc<RwLock<AuthSlots>>,
51}
52
53#[derive(Default)]
54struct AuthSlots {
55    /// session 凭据出站头：`(header_name, header_value)`。`None` = 未登录 / 未注入。
56    session: Option<(String, String)>,
57    /// bot 凭据出站头（如 `("Authorization", "Bearer mp_xxx")`）。
58    bot: Option<(String, String)>,
59}
60
61impl AuthTokenRegistry {
62    pub fn new() -> Self {
63        Self::default()
64    }
65
66    /// 注入 / 覆盖 session 凭据出站头（平台壳登录后调；如 `("cookieId", "<userId>")`）。
67    pub fn set_session(&self, header_name: impl Into<String>, header_value: impl Into<String>) {
68        self.inner.write().session = Some((header_name.into(), header_value.into()));
69    }
70
71    /// 注入 / 覆盖 bot 凭据出站头（如 `("Authorization", "Bearer mp_...")`）。
72    pub fn set_bot(&self, header_name: impl Into<String>, header_value: impl Into<String>) {
73        self.inner.write().bot = Some((header_name.into(), header_value.into()));
74    }
75
76    /// 清空某类凭据（如 401 logout 后）。
77    pub fn clear(&self, kind: AuthKind) {
78        let mut slots = self.inner.write();
79        match kind {
80            AuthKind::Session => slots.session = None,
81            AuthKind::Bot => slots.bot = None,
82        }
83    }
84
85    /// 取某类凭据当前出站头快照（横切层注入用）。
86    fn resolve(&self, kind: AuthKind) -> Option<(String, String)> {
87        let slots = self.inner.read();
88        match kind {
89            AuthKind::Session => slots.session.clone(),
90            AuthKind::Bot => slots.bot.clone(),
91        }
92    }
93}
94
95/// 出站 HTTP 横切装饰器：鉴权注入 + 业务 outcome 观察 + traceparent 透传。
96#[derive(Clone)]
97pub struct CrossCuttingHttp<H, E> {
98    inner: H,
99    event_sink: Arc<E>,
100    auth: AuthTokenRegistry,
101    outcome_observer: HttpOutcomeObserver,
102    otel: Option<HostOtelRuntime>,
103}
104
105impl<H, E> CrossCuttingHttp<H, E>
106where
107    H: HttpRequester,
108    E: EventSink,
109{
110    pub fn new(
111        inner: H,
112        event_sink: Arc<E>,
113        auth: AuthTokenRegistry,
114        outcome_observer: HttpOutcomeObserver,
115    ) -> Self {
116        Self {
117            inner,
118            event_sink,
119            auth,
120            outcome_observer,
121            otel: None,
122        }
123    }
124
125    pub fn with_otel(mut self, otel: HostOtelRuntime) -> Self {
126        self.otel = Some(otel).filter(HostOtelRuntime::is_enabled);
127        self
128    }
129
130    pub fn is_otel_enabled(&self) -> bool {
131        self.otel.is_some()
132    }
133
134    /// 鉴权头注入意图的纯函数（无 I/O）：读 `X-Auth-Kind` → 注真 token → strip 意图头。
135    /// traceparent 等其余头**原样保留**（透传）。返回改写后的请求 + 解析出的意图（供测试断言）。
136    ///
137    /// 意图头缺失 / 未知值 → 不注入（零信任不臆测），仅 strip 任何残留意图头。
138    fn apply_auth(&self, req: HttpRequest) -> (HttpRequest, Option<AuthKind>) {
139        apply_auth_intent(req, |kind| self.auth.resolve(kind))
140    }
141}
142
143#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)]
144#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))]
145impl<H, E> HttpRequester for CrossCuttingHttp<H, E>
146where
147    H: HttpRequester,
148    E: EventSink,
149{
150    /// 执行鉴权后的 HTTP 请求，并把 response/error 作为 request span 的显式子节点。
151    async fn request(&self, req: HttpRequest) -> Result<HttpResponse, PortError> {
152        let (prepared, _kind) = self.apply_auth(req);
153        let url = prepared.url.clone();
154        let carrier = TraceCarrier::from_headers(&prepared.headers);
155        let request_scope = self.otel.as_ref().map(|runtime| {
156            runtime.span_with_attributes(
157                "helix.http.request",
158                TraceDirection::Outbound,
159                carrier.as_ref(),
160                full_debug_http_attributes(runtime, &prepared),
161            )
162        });
163        let outcome = self.inner.request(prepared).await;
164        // 优先使用 Go 响应携带的服务端 Span；没有响应头时仍回退到本地 request child。
165        let response_parent = match &outcome {
166            Ok(response) => TraceCarrier::from_response_headers(&response.headers),
167            Err(_) => None,
168        }
169        .or_else(|| {
170            request_scope
171                .as_ref()
172                .and_then(HostSpanScope::child_carrier)
173        })
174        .or_else(|| carrier.clone());
175        let _response_scope = match &outcome {
176            Ok(response) => self.otel.as_ref().map(|runtime| {
177                let mut attributes =
178                    vec![("http.response.status_code", response.status.to_string())];
179                if runtime.is_full_debug() {
180                    attributes.push((
181                        "helix.debug.http.response_headers",
182                        bounded_debug_string(
183                            &serde_json::to_string(&response.headers)
184                                .unwrap_or_else(|_| "[]".to_string()),
185                        ),
186                    ));
187                    attributes.push((
188                        "helix.debug.http.response_body_base64",
189                        bounded_debug_bytes(&response.body),
190                    ));
191                }
192                runtime.span_with_attributes(
193                    "helix.http.response",
194                    TraceDirection::Inbound,
195                    response_parent.as_ref(),
196                    attributes,
197                )
198            }),
199            Err(error) => self.otel.as_ref().map(|runtime| {
200                runtime.span_with_attributes(
201                    "helix.http.response",
202                    TraceDirection::Inbound,
203                    response_parent.as_ref(),
204                    vec![("helix.http.error_kind", error_kind(error).to_string())],
205                )
206            }),
207        };
208        if let Some(event) = (self.outcome_observer)(&url, &outcome) {
209            self.event_sink.emit(event);
210        }
211        outcome
212    }
213}
214
215/// 将 HTTP 失败折叠成低基数错误类别，不把响应正文或凭据写入 Trace。
216fn error_kind(error: &PortError) -> &'static str {
217    match error {
218        PortError::Storage(_) => "storage",
219        PortError::StorageConflict => "storage_conflict",
220        PortError::Transport(_) => "transport",
221        PortError::Http(_) => "http",
222        PortError::Clock(_) => "clock",
223        PortError::IdSource(_) => "id_source",
224        PortError::Other(_) => "other",
225    }
226}
227
228const FULL_DEBUG_PAYLOAD_LIMIT: usize = 64 * 1024;
229
230/// safe 模式返回空属性；full_debug 才将鉴权后请求送入受控 OTLP span。
231fn full_debug_http_attributes(
232    runtime: &HostOtelRuntime,
233    request: &HttpRequest,
234) -> Vec<(&'static str, String)> {
235    if !runtime.is_full_debug() {
236        return Vec::new();
237    }
238    let mut attributes = vec![
239        (
240            "helix.debug.http.request_headers",
241            bounded_debug_string(
242                &serde_json::to_string(&request.headers).unwrap_or_else(|_| "[]".to_string()),
243            ),
244        ),
245        (
246            "helix.debug.http.request_url",
247            bounded_debug_string(&request.url),
248        ),
249        (
250            "helix.debug.http.request_method",
251            bounded_debug_string(&request.method),
252        ),
253    ];
254    if let Some(body) = &request.body {
255        attributes.push((
256            "helix.debug.http.request_body_base64",
257            bounded_debug_bytes(body),
258        ));
259    }
260    attributes
261}
262
263/// 限制 full_debug 文本长度，避免一个异常请求撑爆 exporter queue。
264fn bounded_debug_string(value: &str) -> String {
265    if value.len() <= FULL_DEBUG_PAYLOAD_LIMIT {
266        return value.to_string();
267    }
268    let boundary = value
269        .char_indices()
270        .map(|(index, _)| index)
271        .take_while(|index| *index <= FULL_DEBUG_PAYLOAD_LIMIT)
272        .last()
273        .unwrap_or(0);
274    format!(
275        "{}...[truncated {} bytes]",
276        &value[..boundary],
277        value.len() - boundary
278    )
279}
280
281/// 将二进制响应限制并编码为可携带的受控 Debug 属性。
282fn bounded_debug_bytes(bytes: &[u8]) -> String {
283    let shown = bytes.len().min(FULL_DEBUG_PAYLOAD_LIMIT);
284    let mut value = base64_encode(&bytes[..shown]);
285    if shown < bytes.len() {
286        value.push_str(&format!("...[truncated {} bytes]", bytes.len() - shown));
287    }
288    value
289}
helix_driver_host/http_cross.rs

helix_driver_host/
http_cross.rs
