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(),
}
}
}