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
use std::{
net::{IpAddr, SocketAddr},
str::{from_utf8, from_utf8_unchecked},
};
use rusty_ulid::Ulid;
use crate::{
pool::Checkout,
protocol::http::{parser::compare_no_case, GenericHttpStream, Method},
Protocol,
};
use sozu_command_lib::logging::LogContext;
/// This is the container used to store and use information about the session from within a Kawa parser callback
#[derive(Debug)]
pub struct HttpContext {
// ========== Write only
/// set to false if Kawa finds a "Connection" header with a "close" value in the response
pub keep_alive_backend: bool,
/// set to false if Kawa finds a "Connection" header with a "close" value in the request
pub keep_alive_frontend: bool,
/// the value of the sticky session cookie in the request
pub sticky_session_found: Option<String>,
// ---------- Status Line
/// the value of the method in the request line
pub method: Option<Method>,
/// the value of the authority of the request (in the request line of "Host" header)
pub authority: Option<String>,
/// the value of the path in the request line
pub path: Option<String>,
/// the value of the status code in the response line
pub status: Option<u16>,
/// the value of the reason in the response line
pub reason: Option<String>,
// ---------- Additional optional data
pub user_agent: Option<String>,
// ========== Read only
/// signals wether Kawa should write a "Connection" header with a "close" value (request and response)
pub closing: bool,
/// the value of the custom header, named "Sozu-Id", that Kawa should write (request and response)
pub id: Ulid,
pub backend_id: Option<String>,
pub cluster_id: Option<String>,
/// the value of the protocol Kawa should write in the Forwarded headers of the request
pub protocol: Protocol,
/// the value of the public address Kawa should write in the Forwarded headers of the request
pub public_address: SocketAddr,
/// the value of the session address Kawa should write in the Forwarded headers of the request
pub session_address: Option<SocketAddr>,
/// the name of the cookie Kawa should read from the request to get the sticky session
pub sticky_name: String,
/// the sticky session that should be used
/// used to create a "Set-Cookie" header in the response in case it differs from sticky_session_found
pub sticky_session: Option<String>,
}
impl kawa::h1::ParserCallbacks<Checkout> for HttpContext {
fn on_headers(&mut self, stream: &mut GenericHttpStream) {
match stream.kind {
kawa::Kind::Request => self.on_request_headers(stream),
kawa::Kind::Response => self.on_response_headers(stream),
}
}
}
impl HttpContext {
/// Callback for request:
///
/// - edit headers (connection, forwarded, sticky cookie, sozu-id)
/// - save information:
/// - method
/// - authority
/// - path
/// - front keep-alive
/// - sticky cookie
/// - user-agent
fn on_request_headers(&mut self, request: &mut GenericHttpStream) {
let buf = &mut request.storage.mut_buffer();
// Captures the request line
if let kawa::StatusLine::Request {
method,
authority,
path,
..
} = &request.detached.status_line
{
self.method = method.data_opt(buf).map(Method::new);
self.authority = authority
.data_opt(buf)
.and_then(|data| from_utf8(data).ok())
.map(ToOwned::to_owned);
self.path = path
.data_opt(buf)
.and_then(|data| from_utf8(data).ok())
.map(ToOwned::to_owned);
}
// if self.method == Some(Method::Get) && request.body_size == kawa::BodySize::Empty {
// request.parsing_phase = kawa::ParsingPhase::Terminated;
// }
let public_ip = self.public_address.ip();
let public_port = self.public_address.port();
let proto = match self.protocol {
Protocol::HTTP => "http",
Protocol::HTTPS => "https",
_ => unreachable!(),
};
// Find and remove the sticky_name cookie
// if found its value is stored in sticky_session_found
for cookie in &mut request.detached.jar {
let key = cookie.key.data(buf);
if key == self.sticky_name.as_bytes() {
let val = cookie.val.data(buf);
self.sticky_session_found = from_utf8(val).ok().map(|val| val.to_string());
cookie.elide();
}
}
// If found:
// - set Connection to "close" if closing is set
// - set keep_alive_frontend to false if Connection is "close"
// - update value of X-Forwarded-Proto
// - update value of X-Forwarded-Port
// - store X-Forwarded-For
// - store Forwarded
// - store User-Agent
let mut x_for = None;
let mut forwarded = None;
let mut has_x_port = false;
let mut has_x_proto = false;
let mut has_connection = false;
for block in &mut request.blocks {
match block {
kawa::Block::Header(header) if !header.is_elided() => {
let key = header.key.data(buf);
if compare_no_case(key, b"connection") {
has_connection = true;
if self.closing {
header.val = kawa::Store::Static(b"close");
} else {
let val = header.val.data(buf);
self.keep_alive_frontend &= !compare_no_case(val, b"close");
}
} else if compare_no_case(key, b"X-Forwarded-Proto") {
has_x_proto = true;
// header.val = kawa::Store::Static(proto.as_bytes());
incr!("http.trusting.x_proto");
let val = header.val.data(buf);
if !compare_no_case(val, proto.as_bytes()) {
incr!("http.trusting.x_proto.diff");
debug!(
"Trusting X-Forwarded-Proto for {:?} even though {:?} != {}",
self.authority, val, proto
);
}
} else if compare_no_case(key, b"X-Forwarded-Port") {
has_x_port = true;
// header.val = kawa::Store::from_string(public_port.to_string());
incr!("http.trusting.x_port");
let val = header.val.data(buf);
let expected = public_port.to_string();
if !compare_no_case(val, expected.as_bytes()) {
incr!("http.trusting.x_port.diff");
debug!(
"Trusting X-Forwarded-Port for {:?} even though {:?} != {}",
self.authority, val, expected
);
}
} else if compare_no_case(key, b"X-Forwarded-For") {
x_for = Some(header);
} else if compare_no_case(key, b"Forwarded") {
forwarded = Some(header);
} else if compare_no_case(key, b"User-Agent") {
self.user_agent = header
.val
.data_opt(buf)
.and_then(|data| from_utf8(data).ok())
.map(ToOwned::to_owned);
}
}
_ => {}
}
}
// If session_address is set:
// - append its ip address to the list of "X-Forwarded-For" if it was found, creates it if not
// - append "proto=[PROTO];for=[PEER];by=[PUBLIC]" to the list of "Forwarded" if it was found, creates it if not
if let Some(peer_addr) = self.session_address {
let peer_ip = peer_addr.ip();
let peer_port = peer_addr.port();
let has_x_for = x_for.is_some();
let has_forwarded = forwarded.is_some();
if let Some(header) = x_for {
header.val = kawa::Store::from_string(format!("{}, {peer_ip}", unsafe {
from_utf8_unchecked(header.val.data(buf))
}));
}
if let Some(header) = &mut forwarded {
let value = unsafe { from_utf8_unchecked(header.val.data(buf)) };
let new_value = match (peer_ip, public_ip) {
(IpAddr::V4(_), IpAddr::V4(_)) => {
format!("{value}, proto={proto};for={peer_ip}:{peer_port};by={public_ip}")
}
(IpAddr::V4(_), IpAddr::V6(_)) => {
format!(
"{value}, proto={proto};for={peer_ip}:{peer_port};by=\"{public_ip}\""
)
}
(IpAddr::V6(_), IpAddr::V4(_)) => {
format!(
"{value}, proto={proto};for=\"{peer_ip}:{peer_port}\";by={public_ip}"
)
}
(IpAddr::V6(_), IpAddr::V6(_)) => {
format!(
"{value}, proto={proto};for=\"{peer_ip}:{peer_port}\";by=\"{public_ip}\""
)
}
};
header.val = kawa::Store::from_string(new_value);
}
if !has_x_for {
request.push_block(kawa::Block::Header(kawa::Pair {
key: kawa::Store::Static(b"X-Forwarded-For"),
val: kawa::Store::from_string(peer_ip.to_string()),
}));
}
if !has_forwarded {
let value = match (peer_ip, public_ip) {
(IpAddr::V4(_), IpAddr::V4(_)) => {
format!("proto={proto};for={peer_ip}:{peer_port};by={public_ip}")
}
(IpAddr::V4(_), IpAddr::V6(_)) => {
format!("proto={proto};for={peer_ip}:{peer_port};by=\"{public_ip}\"")
}
(IpAddr::V6(_), IpAddr::V4(_)) => {
format!("proto={proto};for=\"{peer_ip}:{peer_port}\";by={public_ip}")
}
(IpAddr::V6(_), IpAddr::V6(_)) => {
format!("proto={proto};for=\"{peer_ip}:{peer_port}\";by=\"{public_ip}\"")
}
};
request.push_block(kawa::Block::Header(kawa::Pair {
key: kawa::Store::Static(b"Forwarded"),
val: kawa::Store::from_string(value),
}));
}
}
if !has_x_port {
request.push_block(kawa::Block::Header(kawa::Pair {
key: kawa::Store::Static(b"X-Forwarded-Port"),
val: kawa::Store::from_string(public_port.to_string()),
}));
}
if !has_x_proto {
request.push_block(kawa::Block::Header(kawa::Pair {
key: kawa::Store::Static(b"X-Forwarded-Proto"),
val: kawa::Store::Static(proto.as_bytes()),
}));
}
// Create a "Connection" header in case it was not found and closing it set
if !has_connection && self.closing {
request.push_block(kawa::Block::Header(kawa::Pair {
key: kawa::Store::Static(b"Connection"),
val: kawa::Store::Static(b"close"),
}));
}
// Create a custom "Sozu-Id" header
request.push_block(kawa::Block::Header(kawa::Pair {
key: kawa::Store::Static(b"Sozu-Id"),
val: kawa::Store::from_string(self.id.to_string()),
}));
}
/// Callback for response:
///
/// - edit headers (connection, set-cookie, sozu-id)
/// - save information:
/// - status code
/// - reason
/// - back keep-alive
fn on_response_headers(&mut self, response: &mut GenericHttpStream) {
let buf = &mut response.storage.mut_buffer();
// Captures the response line
if let kawa::StatusLine::Response { code, reason, .. } = &response.detached.status_line {
self.status = Some(*code);
self.reason = reason
.data_opt(buf)
.and_then(|data| from_utf8(data).ok())
.map(ToOwned::to_owned);
}
if self.method == Some(Method::Head) {
response.parsing_phase = kawa::ParsingPhase::Terminated;
}
// If found:
// - set Connection to "close" if closing is set
// - set keep_alive_backend to false if Connection is "close"
for block in &mut response.blocks {
match block {
kawa::Block::Header(header) if !header.is_elided() => {
let key = header.key.data(buf);
if compare_no_case(key, b"connection") {
if self.closing {
header.val = kawa::Store::Static(b"close");
} else {
let val = header.val.data(buf);
self.keep_alive_backend &= !compare_no_case(val, b"close");
}
}
}
_ => {}
}
}
// If the sticky_session is set and differs from the one found in the request
// create a "Set-Cookie" header to update the sticky_name value
if let Some(sticky_session) = &self.sticky_session {
if self.sticky_session != self.sticky_session_found {
response.push_block(kawa::Block::Header(kawa::Pair {
key: kawa::Store::Static(b"Set-Cookie"),
val: kawa::Store::from_string(format!(
"{}={}; Path=/",
self.sticky_name, sticky_session
)),
}));
}
}
// Create a custom "Sozu-Id" header
response.push_block(kawa::Block::Header(kawa::Pair {
key: kawa::Store::Static(b"Sozu-Id"),
val: kawa::Store::from_string(self.id.to_string()),
}));
}
pub fn reset(&mut self) {
self.keep_alive_backend = true;
self.keep_alive_frontend = true;
self.sticky_session_found = None;
self.method = None;
self.authority = None;
self.path = None;
self.status = None;
self.reason = None;
self.user_agent = None;
}
pub fn log_context(&self) -> LogContext {
LogContext {
request_id: self.id,
cluster_id: self.cluster_id.as_deref(),
backend_id: self.backend_id.as_deref(),
}
}
}