zenith-http1 0.1.0

Zenith HTTP/1.1 协议解析器(RFC 7230):零堆分配热路径、流式解析、CRLF 防注入
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
//! HTTP/1.1 连接管理
//!
//! 管理单连接上的请求-响应循环(pipelining 支持可选):
//! - 状态机:Waiting → Reading → Processing → Sending → Waiting/Closed
//! - keep-alive 超时
//! - 最大请求数限制
//! - 优雅关闭

use crate::parser::Http1Parser;
use crate::types::{Http1Config, Http1Error, HttpRequest};

/// HTTP/1.1 连接状态
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Http1ConnectionState {
    /// 等待请求
    Waiting,
    /// 读取请求
    ReadingRequest,
    /// 读取头部
    ReadingHeaders,
    /// 读取请求体
    ReadingBody,
    /// 处理请求
    Processing,
    /// 发送响应
    SendingResponse,
    /// 关闭中
    Closing,
    /// 已关闭
    Closed,
}

/// HTTP/1.1 连接
#[derive(Debug)]
pub struct Http1Connection {
    state: Http1ConnectionState,
    parser: Http1Parser,
    /// 当前请求
    current_request: Option<HttpRequest>,
    /// 是否启用 keep-alive
    keep_alive: bool,
    /// 已处理请求数
    requests_handled: u64,
    /// 是否关闭
    closed: bool,
    /// 上次活动时间戳(毫秒)
    last_activity_ms: u64,
    /// update_time 是否已被调用(区分"未初始化"与"update_time(0)")
    time_initialized: bool,
}

impl Http1Connection {
    /// 创建连接
    #[inline]
    pub fn new(config: Http1Config) -> Self {
        Self {
            keep_alive: config.keep_alive,
            parser: Http1Parser::new(config),
            state: Http1ConnectionState::Waiting,
            current_request: None,
            requests_handled: 0,
            closed: false,
            last_activity_ms: 0,
            time_initialized: false,
        }
    }

    /// 当前状态
    #[inline]
    pub fn state(&self) -> Http1ConnectionState {
        self.state
    }

    /// 是否关闭
    #[inline]
    pub fn is_closed(&self) -> bool {
        self.closed
    }

    /// 输入数据,尝试解析请求
    ///
    /// # 返回
    /// - Ok(Some(req, body_consumed)): 已解析完整请求,可能包含 body 字节消耗
    /// - Ok((None, consumed)): 需要更多数据
    /// - Err(err): 协议错误(错误会自动关闭连接)
    pub fn on_data(&mut self, data: &[u8]) -> Result<(Option<HttpRequest>, usize), Http1Error> {
        if self.closed {
            return Err(Http1Error::ConnectionClosed);
        }

        // Slowloris 防护:每次收到数据刷新活跃时间戳
        // (含 body/chunked 阶段,M-9:body 中途 idle 超时同样生效)
        // 仅在 update_time 已初始化时间戳时刷新 parser 计时,
        // 否则跳过以避免 first_byte_ms 被误设为 0(struct 默认值)
        if !data.is_empty() && self.time_initialized {
            self.parser.note_activity(self.last_activity_ms);
        }

        let (req, consumed) = match self.parser.feed(data) {
            Ok(v) => v,
            Err(e) => {
                self.on_error(&e);
                return Err(e);
            }
        };

        if let Some(mut req) = req {
            // 检查 Connection: close
            if !req.keep_alive {
                self.keep_alive = false;
            }
            // Http1Parser 现已内置完整 chunked 请求体解析(RFC 7230 §4.1),
            // 当 req.chunked == true 时 req.body 已包含解码后的完整 body,
            // 无需上层通过 body_complete() 推送。
            // - Content-Length > 0 且 req.body 非空:Http1Parser 已同步读取 body,直接进入 Processing
            // - Content-Length > 0 且 req.body 为空:兼容模式,进入 ReadingBody 等待外部 body_complete()
            // - Content-Length == 0 或 None:无 body,直接处理
            if req.chunked {
                // chunked body 已由 parser 完整解码,直接进入 Processing
                self.state = Http1ConnectionState::Processing;
            } else if let Some(len) = req.content_length {
                if len == 0 {
                    self.state = Http1ConnectionState::Processing;
                } else if !req.body.is_empty() && req.body.len() as u64 == len {
                    // Http1Parser 已按 Content-Length 完整读取 body,无需再等待
                    self.state = Http1ConnectionState::Processing;
                } else {
                    // 兼容旧语义:parser 未读取 body,等待上层调用 body_complete()
                    self.state = Http1ConnectionState::ReadingBody;
                }
            } else if req.line.method.as_ref() == "POST" || req.line.method.as_ref() == "PUT" {
                // 方法无 content-length 时视为 body 长度为 0
                req.content_length = Some(0);
                self.state = Http1ConnectionState::Processing;
            } else {
                self.state = Http1ConnectionState::Processing;
            }
            self.current_request = Some(req);
            self.requests_handled += 1;
            Ok((self.current_request.clone(), consumed))
        } else {
            // parser 尚未返回完整请求(仍在读取 header / body / chunked 帧)
            // 连接层状态反映 parser 的活跃状态
            self.state = match self.parser.state() {
                crate::parser::ParserState::WaitingRequest
                | crate::parser::ParserState::ReadingRequest => Http1ConnectionState::ReadingRequest,
                crate::parser::ParserState::ReadingHeaders => Http1ConnectionState::ReadingHeaders,
                crate::parser::ParserState::ReadingBody
                | crate::parser::ParserState::ReadingChunkSize
                | crate::parser::ParserState::ReadingChunkData
                | crate::parser::ParserState::ReadingChunkTrailer => Http1ConnectionState::ReadingBody,
                crate::parser::ParserState::HeadersComplete => Http1ConnectionState::Processing,
                crate::parser::ParserState::Error => Http1ConnectionState::Closed,
            };
            Ok((None, consumed))
        }
    }

    /// 设置请求体已接收完成
    pub fn body_complete(&mut self, body: Vec<u8>) {
        if let Some(req) = self.current_request.as_mut() {
            req.body = body;
            self.state = Http1ConnectionState::Processing;
        }
    }

    /// 获取当前请求
    #[inline]
    pub fn current_request(&self) -> Option<&HttpRequest> {
        self.current_request.as_ref()
    }

    /// 响应发送完成
    pub fn response_sent(&mut self) {
        self.current_request = None;
        if !self.keep_alive {
            self.state = Http1ConnectionState::Closing;
            self.closed = true;
        } else {
            // 准备下一个请求
            self.parser.reset();
            self.state = Http1ConnectionState::Waiting;
        }
    }

    /// 处理错误:关闭连接
    pub fn on_error(&mut self, _err: &Http1Error) {
        self.state = Http1ConnectionState::Closing;
        self.closed = true;
    }

    /// 获取已处理请求数
    #[inline]
    pub fn requests_handled(&self) -> u64 {
        self.requests_handled
    }

    /// 更新当前时间戳(由事件循环每周期调用)
    ///
    /// 必须在首次 [`Self::on_data`] 之前调用至少一次,否则 parser 的
    /// `first_byte_ms` 不会被设置(note_activity 被跳过),Slowloris
    /// 空闲超时检测在首字节阶段不生效。调用后 `time_initialized` 置 true,
    /// 后续 `on_data` 才会向 parser 刷新活跃时间戳。
    #[inline]
    pub fn update_time(&mut self, now_ms: u64) {
        self.last_activity_ms = now_ms;
        self.time_initialized = true;
    }

    /// 检查空闲超时(Slowloris 防护)
    ///
    /// 由事件循环在每个周期调用。如果请求解析超时,关闭连接。
    pub fn check_timeout(&mut self) -> Result<(), Http1Error> {
        if self.closed {
            return Ok(());
        }
        if let Err(e) = self.parser.check_idle_timeout(self.last_activity_ms) {
            self.on_error(&e);
            return Err(e);
        }
        Ok(())
    }

    /// 关闭连接
    pub fn close(&mut self) {
        self.state = Http1ConnectionState::Closed;
        self.closed = true;
    }
}

impl Default for Http1Connection {
    fn default() -> Self {
        Self::new(Http1Config::new())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_connection_basic() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let input = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n";
        let (req, _) = conn.on_data(input).unwrap();
        assert!(req.is_some());
        assert_eq!(conn.state(), Http1ConnectionState::Processing);
    }

    #[test]
    fn test_connection_lifecycle() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let input = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n";
        conn.on_data(input).unwrap();
        assert_eq!(conn.state(), Http1ConnectionState::Processing);

        conn.response_sent();
        assert_eq!(conn.state(), Http1ConnectionState::Waiting);
        assert!(!conn.is_closed());
    }

    #[test]
    fn test_connection_close() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let input =
            b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n";
        conn.on_data(input).unwrap();
        conn.response_sent();
        assert!(conn.is_closed());
    }

    #[test]
    fn test_connection_error_cases() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let r = conn.on_data(b"INV HTTP/1.1\r\nHost: x\r\n\r\n");
        assert!(r.is_err());
        assert!(conn.is_closed());
    }

    #[test]
    fn test_connection_body_post() {
        let mut conn = Http1Connection::new(Http1Config::new());
        // Http1Parser 在 Content-Length 模式下会同步读取完整 body,因此此处需要同时提供 headers + body
        //(不再走 ReadingBody → body_complete 路径,而是直接进入 Processing)
        let input =
            b"POST /api HTTP/1.1\r\nHost: example.com\r\nContent-Length: 5\r\n\r\nhello";
        let (req, _) = conn.on_data(input).unwrap();
        let req = req.unwrap();
        assert_eq!(req.content_length, Some(5));
        assert_eq!(req.body, b"hello");
        assert_eq!(conn.state(), Http1ConnectionState::Processing);
    }

    #[test]
    fn test_connection_state_variants() {
        let states = [
            Http1ConnectionState::Waiting,
            Http1ConnectionState::ReadingRequest,
            Http1ConnectionState::ReadingHeaders,
            Http1ConnectionState::ReadingBody,
            Http1ConnectionState::Processing,
            Http1ConnectionState::SendingResponse,
            Http1ConnectionState::Closing,
            Http1ConnectionState::Closed,
        ];
        for (i, s) in states.iter().enumerate() {
            assert_eq!(*s, states[i]);
        }
        assert_ne!(Http1ConnectionState::Waiting, Http1ConnectionState::Closed);
    }

    #[test]
    fn test_connection_default() {
        let conn = Http1Connection::default();
        assert_eq!(conn.state(), Http1ConnectionState::Waiting);
        assert!(!conn.is_closed());
        assert_eq!(conn.requests_handled(), 0);
    }

    #[test]
    fn test_connection_keep_alive_multiple_requests() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let input1 = b"GET /1 HTTP/1.1\r\nHost: example.com\r\n\r\n";
        let (req1, _) = conn.on_data(input1).unwrap();
        assert!(req1.is_some());
        assert_eq!(conn.requests_handled(), 1);

        conn.response_sent();
        assert_eq!(conn.state(), Http1ConnectionState::Waiting);

        let input2 = b"GET /2 HTTP/1.1\r\nHost: example.com\r\n\r\n";
        let (req2, _) = conn.on_data(input2).unwrap();
        assert!(req2.is_some());
        assert_eq!(conn.requests_handled(), 2);
    }

    #[test]
    fn test_connection_closed_rejects_data() {
        let mut conn = Http1Connection::new(Http1Config::new());
        conn.close();
        assert!(conn.is_closed());
        assert_eq!(conn.state(), Http1ConnectionState::Closed);

        let input = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n";
        let r = conn.on_data(input);
        assert!(r.is_err());
        assert!(matches!(r.unwrap_err(), Http1Error::ConnectionClosed));
    }

    #[test]
    fn test_connection_current_request() {
        let mut conn = Http1Connection::new(Http1Config::new());
        assert!(conn.current_request().is_none());

        let input = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n";
        conn.on_data(input).unwrap();
        assert!(conn.current_request().is_some());
        assert_eq!(conn.current_request().unwrap().line.method.as_ref(), "GET");
    }

    #[test]
    fn test_connection_chunked_body_state() {
        let mut conn = Http1Connection::new(Http1Config::new());
        // 完整 chunked 请求(RFC 7230 §4.1):headers + chunk + last-chunk + trailer CRLF
        let input = b"POST /upload HTTP/1.1\r\nHost: example.com\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nHello\r\n0\r\n\r\n";
        let (req, _) = conn.on_data(input).unwrap();
        assert!(req.is_some());
        let req = req.unwrap();
        assert_eq!(conn.state(), Http1ConnectionState::Processing);
        assert_eq!(req.body, b"Hello");
    }

    #[test]
    fn test_connection_chunked_body_partial() {
        let mut conn = Http1Connection::new(Http1Config::new());
        // 仅 headers,未包含 chunk 数据 → parser 返回 None,连接进入 ReadingBody
        let part1 = b"POST /upload HTTP/1.1\r\nHost: example.com\r\nTransfer-Encoding: chunked\r\n\r\n";
        let (req, _) = conn.on_data(part1).unwrap();
        assert!(req.is_none());
        assert_eq!(conn.state(), Http1ConnectionState::ReadingBody);

        // 后续 chunk 数据到达 → parser 完成解析,连接进入 Processing
        let part2 = b"5\r\nHello\r\n0\r\n\r\n";
        let (req, _) = conn.on_data(part2).unwrap();
        assert!(req.is_some());
        assert_eq!(conn.state(), Http1ConnectionState::Processing);
        assert_eq!(req.unwrap().body, b"Hello");
    }

    #[test]
    fn test_connection_zero_content_length() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let input = b"POST /api HTTP/1.1\r\nHost: example.com\r\nContent-Length: 0\r\n\r\n";
        let (req, _) = conn.on_data(input).unwrap();
        assert!(req.is_some());
        assert_eq!(conn.state(), Http1ConnectionState::Processing);
    }

    #[test]
    fn test_connection_put_method() {
        let mut conn = Http1Connection::new(Http1Config::new());
        // Http1Parser 同步读取 Content-Length body,提供完整 headers+body 直接进入 Processing
        let input =
            b"PUT /resource HTTP/1.1\r\nHost: example.com\r\nContent-Length: 5\r\n\r\nworld";
        let (req, _) = conn.on_data(input).unwrap();
        let req = req.unwrap();
        assert_eq!(req.line.method.as_ref(), "PUT");
        assert_eq!(req.body, b"world");
        assert_eq!(conn.state(), Http1ConnectionState::Processing);
    }

    #[test]
    fn test_connection_head_method() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let input = b"HEAD / HTTP/1.1\r\nHost: example.com\r\n\r\n";
        let (req, _) = conn.on_data(input).unwrap();
        assert!(req.is_some());
        assert_eq!(req.unwrap().line.method.as_ref(), "HEAD");
        assert_eq!(conn.state(), Http1ConnectionState::Processing);
    }

    #[test]
    fn test_connection_on_error_closes_connection() {
        let mut conn = Http1Connection::new(Http1Config::new());
        conn.on_error(&Http1Error::SyntaxError("test".into()));
        assert!(conn.is_closed());
        assert_eq!(conn.state(), Http1ConnectionState::Closing);
    }

    #[test]
    fn test_connection_body_complete_no_current_request() {
        let mut conn = Http1Connection::new(Http1Config::new());
        conn.body_complete(b"test".to_vec());
        assert_eq!(conn.state(), Http1ConnectionState::Waiting);
    }

    #[test]
    fn test_connection_response_sent_clears_request() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let input = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n";
        conn.on_data(input).unwrap();
        assert!(conn.current_request().is_some());
        conn.response_sent();
        assert!(conn.current_request().is_none());
    }

    #[test]
    fn test_connection_closing_state() {
        let mut conn = Http1Connection::new(Http1Config::new());
        let input = b"GET / HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n\r\n";
        conn.on_data(input).unwrap();
        conn.response_sent();
        assert_eq!(conn.state(), Http1ConnectionState::Closing);
        assert!(conn.is_closed());
    }

    #[test]
    fn test_connection_debug_format() {
        let conn = Http1Connection::new(Http1Config::new());
        let s = format!("{:?}", conn);
        assert!(!s.is_empty());
    }

    #[test]
    fn test_connection_idle_timeout_during_body_read() {
        // M-9:body 中途静默超过 idle_timeout → 连接层 IdleTimeout 并关闭
        let config = Http1Config::new().with_idle_timeout_ms(30_000);
        let mut conn = Http1Connection::new(config);
        conn.update_time(0);
        // 仅头部到达(CL=10,body 未到)→ ReadingBody
        let (req, _) = conn
            .on_data(b"POST / HTTP/1.1\r\nHost: example.com\r\nContent-Length: 10\r\n\r\n")
            .unwrap();
        assert!(req.is_none());
        assert_eq!(conn.state(), Http1ConnectionState::ReadingBody);
        // 30s+ 未再收到 body → IdleTimeout,连接 fail-closed
        conn.update_time(30_001);
        let r = conn.check_timeout();
        assert!(matches!(r, Err(Http1Error::IdleTimeout)), "实际 {r:?}");
        assert!(conn.is_closed());
    }

    #[test]
    fn test_connection_body_activity_prevents_timeout() {
        // M-9 对照:body 分片持续到达刷新活跃,不得误判超时
        let config = Http1Config::new().with_idle_timeout_ms(30_000);
        let mut conn = Http1Connection::new(config);
        conn.update_time(0);
        let (req, _) = conn
            .on_data(b"POST / HTTP/1.1\r\nHost: example.com\r\nContent-Length: 6\r\n\r\nhe")
            .unwrap();
        assert!(req.is_none());
        // 20s 后 body 第二片到达
        conn.update_time(20_000);
        let (req, _) = conn.on_data(b"ll").unwrap();
        assert!(req.is_none());
        // 活跃后 25s:不得超时
        conn.update_time(45_000);
        assert!(conn.check_timeout().is_ok(), "活跃刷新后不得超时");
        // 最后一片凑满 6 字节 body 完成请求
        let (req, _) = conn.on_data(b"lo").unwrap();
        assert!(req.is_some());
    }
}