mirrors/actix-web
1
0
Fork 0
mirror of https://github.com/actix/actix-web.git synced 2024-11-20 08:31:09 +00:00
Code Issues Projects Releases Wiki Activity

Expose peer addr via Request::peer_addr() and RequestHead::peer_addr

Browse source
This commit is contained in:
Nikolay Kim 2019-04-16 09:54:02 -07:00
parent 14252f5ef2
commit a116c4c2c7
13 changed files with 170 additions and 85 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
Cargo.toml
View file

@ -72,8 +72,8 @@ actix-router = "0.1.2"
actix-rt = "0.2.2"
actix-web-codegen = "0.1.0-alpha.6"
actix-http = { version = "0.1.0-alpha.5", features=["fail"] }
actix-server = "0.4.2"
actix-server-config = "0.1.0"
actix-server = "0.4.3"
actix-server-config = "0.1.1"
actix-threadpool = "0.1.0"
awc = { version = "0.1.0-alpha.6", optional = true }
@ -81,7 +81,7 @@ bytes = "0.4"
derive_more = "0.14"
encoding = "0.2"
futures = "0.1"
hashbrown = "0.2.1"
hashbrown = "0.2.2"
log = "0.4"
mime = "0.3"
net2 = "0.2.33"
@ -121,4 +121,4 @@ actix-web-codegen = { path = "actix-web-codegen" }
actix-web-actors = { path = "actix-web-actors" }
actix-session = { path = "actix-session" }
actix-files = { path = "actix-files" }
awc = { path = "awc" }
awc = { path = "awc" }

6
actix-http/CHANGES.md
View file

@ -1,6 +1,10 @@
# Changes
## [0.1.0] - 2019-04-xx
## [0.1.0] - 2019-04-16
### Added
* Expose peer addr via `Request::peer_addr()` and `RequestHead::peer_addr`
### Changed

14
actix-http/Cargo.toml
View file

@ -1,6 +1,6 @@
[package]
name = "actix-http"
version = "0.1.0-alpha.5"
version = "0.1.0"
authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
description = "Actix http primitives"
readme = "README.md"
@ -26,7 +26,7 @@ path = "src/lib.rs"
default = []
# openssl
ssl = ["openssl", "actix-connect/ssl"]
ssl = ["openssl", "actix-connect/ssl", "actix-server-config/ssl"]
# brotli encoding, requires c compiler
brotli = ["brotli2"]
@ -48,7 +48,7 @@ actix-service = "0.3.6"
actix-codec = "0.1.2"
actix-connect = "0.1.4"
actix-utils = "0.3.5"
actix-server-config = "0.1.0"
actix-server-config = "0.1.1"
actix-threadpool = "0.1.0"
base64 = "0.10"
@ -60,7 +60,7 @@ derive_more = "0.14"
either = "1.5.2"
encoding = "0.2"
futures = "0.1"
hashbrown = "0.2.0"
hashbrown = "0.2.2"
h2 = "0.1.16"
http = "0.1.17"
httparse = "1.3"
@ -76,10 +76,10 @@ serde = "1.0"
serde_json = "1.0"
sha1 = "0.6"
slab = "0.4"
serde_urlencoded = "0.5.3"
serde_urlencoded = "0.5.5"
time = "0.1"
tokio-tcp = "0.1.3"
tokio-timer = "0.2"
tokio-timer = "0.2.8"
tokio-current-thread = "0.1"
trust-dns-resolver = { version="0.11.0", default-features = false }
@ -96,7 +96,7 @@ openssl = { version="0.10", optional = true }
[dev-dependencies]
actix-rt = "0.2.2"
actix-server = { version = "0.4.1", features=["ssl"] }
actix-server = { version = "0.4.3", features=["ssl"] }
actix-connect = { version = "0.1.4", features=["ssl"] }
actix-http-test = { version = "0.1.0-alpha.3", features=["ssl"] }
env_logger = "0.6"

9
actix-http/src/h1/codec.rs
View file

@ -1,6 +1,6 @@
#![allow(unused_imports, unused_variables, dead_code)]
use std::fmt;
use std::io::{self, Write};
use std::io::Write;
use std::{fmt, io, net};
use actix_codec::{Decoder, Encoder};
use bitflags::bitflags;
@ -40,7 +40,6 @@ pub struct Codec {
// encoder part
flags: Flags,
encoder: encoder::MessageEncoder<Response<()>>,
// headers_size: u32,
}
impl Default for Codec {
@ -67,13 +66,11 @@ impl Codec {
};
Codec {
config,
flags,
decoder: decoder::MessageDecoder::default(),
payload: None,
version: Version::HTTP_11,
ctype: ConnectionType::Close,
flags,
// headers_size: 0,
encoder: encoder::MessageEncoder::default(),
}
}

20
actix-http/src/h1/dispatcher.rs
View file

@ -1,8 +1,9 @@
use std::collections::VecDeque;
use std::time::Instant;
use std::{fmt, io};
use std::{fmt, io, net};
use actix_codec::{AsyncRead, AsyncWrite, Decoder, Encoder, Framed, FramedParts};
use actix_codec::{Decoder, Encoder, Framed, FramedParts};
use actix_server_config::IoStream;
use actix_service::Service;
use actix_utils::cloneable::CloneableService;
use bitflags::bitflags;
@ -81,6 +82,7 @@ where
expect: CloneableService<X>,
upgrade: Option<CloneableService<U>>,
flags: Flags,
peer_addr: Option<net::SocketAddr>,
error: Option<DispatchError>,
state: State<S, B, X>,
@ -161,7 +163,7 @@ impl PartialEq for PollResponse {
impl<T, S, B, X, U> Dispatcher<T, S, B, X, U>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: Service<Request = Request>,
S::Error: Into<Error>,
S::Response: Into<Response<B>>,
@ -220,14 +222,15 @@ where
Dispatcher {
inner: DispatcherState::Normal(InnerDispatcher {
io,
codec,
read_buf,
write_buf: BytesMut::with_capacity(HW_BUFFER_SIZE),
payload: None,
state: State::None,
error: None,
peer_addr: io.peer_addr(),
messages: VecDeque::new(),
io,
codec,
read_buf,
service,
expect,
upgrade,
@ -241,7 +244,7 @@ where
impl<T, S, B, X, U> InnerDispatcher<T, S, B, X, U>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: Service<Request = Request>,
S::Error: Into<Error>,
S::Response: Into<Response<B>>,
@ -490,6 +493,7 @@ where
match msg {
Message::Item(mut req) => {
let pl = self.codec.message_type();
req.head_mut().peer_addr = self.peer_addr;
if pl == MessageType::Stream && self.upgrade.is_some() {
self.messages.push_back(DispatcherMessage::Upgrade(req));
@ -649,7 +653,7 @@ where
impl<T, S, B, X, U> Future for Dispatcher<T, S, B, X, U>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: Service<Request = Request>,
S::Error: Into<Error>,
S::Response: Into<Response<B>>,

20
actix-http/src/h1/service.rs
View file

@ -1,8 +1,8 @@
use std::fmt;
use std::marker::PhantomData;
use actix_codec::{AsyncRead, AsyncWrite, Framed};
use actix_server_config::{Io, ServerConfig as SrvConfig};
use actix_codec::Framed;
use actix_server_config::{Io, IoStream, ServerConfig as SrvConfig};
use actix_service::{IntoNewService, NewService, Service};
use actix_utils::cloneable::CloneableService;
use futures::future::{ok, FutureResult};
@ -104,7 +104,7 @@ where
impl<T, P, S, B, X, U> NewService<SrvConfig> for H1Service<T, P, S, B, X, U>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: NewService<SrvConfig, Request = Request>,
S::Error: Into<Error>,
S::Response: Into<Response<B>>,
@ -161,7 +161,7 @@ where
impl<T, P, S, B, X, U> Future for H1ServiceResponse<T, P, S, B, X, U>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: NewService<SrvConfig, Request = Request>,
S::Error: Into<Error>,
S::Response: Into<Response<B>>,
@ -245,7 +245,7 @@ where
impl<T, P, S, B, X, U> Service for H1ServiceHandler<T, P, S, B, X, U>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: Service<Request = Request>,
S::Error: Into<Error>,
S::Response: Into<Response<B>>,
@ -309,7 +309,7 @@ pub struct OneRequest<T, P> {
impl<T, P> OneRequest<T, P>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
{
/// Create new `H1SimpleService` instance.
pub fn new() -> Self {
@ -322,7 +322,7 @@ where
impl<T, P> NewService<SrvConfig> for OneRequest<T, P>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
{
type Request = Io<T, P>;
type Response = (Request, Framed<T, Codec>);
@ -348,7 +348,7 @@ pub struct OneRequestService<T, P> {
impl<T, P> Service for OneRequestService<T, P>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
{
type Request = Io<T, P>;
type Response = (Request, Framed<T, Codec>);
@ -372,14 +372,14 @@ where
#[doc(hidden)]
pub struct OneRequestServiceResponse<T>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
{
framed: Option<Framed<T, Codec>>,
}
impl<T> Future for OneRequestServiceResponse<T>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
{
type Item = (Request, Framed<T, Codec>);
type Error = ParseError;

19
actix-http/src/h2/dispatcher.rs
View file

@ -1,9 +1,10 @@
use std::collections::VecDeque;
use std::marker::PhantomData;
use std::time::Instant;
use std::{fmt, mem};
use std::{fmt, mem, net};
use actix_codec::{AsyncRead, AsyncWrite};
use actix_server_config::IoStream;
use actix_service::Service;
use actix_utils::cloneable::CloneableService;
use bitflags::bitflags;
@ -29,14 +30,11 @@ use crate::response::Response;
const CHUNK_SIZE: usize = 16_384;
/// Dispatcher for HTTP/2 protocol
pub struct Dispatcher<
T: AsyncRead + AsyncWrite,
S: Service<Request = Request>,
B: MessageBody,
> {
pub struct Dispatcher<T: IoStream, S: Service<Request = Request>, B: MessageBody> {
service: CloneableService<S>,
connection: Connection<T, Bytes>,
config: ServiceConfig,
peer_addr: Option<net::SocketAddr>,
ka_expire: Instant,
ka_timer: Option<Delay>,
_t: PhantomData<B>,
@ -44,7 +42,7 @@ pub struct Dispatcher<
impl<T, S, B> Dispatcher<T, S, B>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: Service<Request = Request>,
S::Error: Into<Error>,
S::Future: 'static,
@ -56,6 +54,7 @@ where
connection: Connection<T, Bytes>,
config: ServiceConfig,
timeout: Option<Delay>,
peer_addr: Option<net::SocketAddr>,
) -> Self {
// let keepalive = config.keep_alive_enabled();
// let flags = if keepalive {
@ -76,9 +75,10 @@ where
Dispatcher {
service,
config,
peer_addr,
connection,
ka_expire,
ka_timer,
connection,
_t: PhantomData,
}
}
@ -86,7 +86,7 @@ where
impl<T, S, B> Future for Dispatcher<T, S, B>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: Service<Request = Request>,
S::Error: Into<Error>,
S::Future: 'static,
@ -117,6 +117,7 @@ where
head.method = parts.method;
head.version = parts.version;
head.headers = parts.headers.into();
head.peer_addr = self.peer_addr;
tokio_current_thread::spawn(ServiceResponse::<S::Future, B> {
state: ServiceResponseState::ServiceCall(
self.service.call(req),

58
actix-http/src/h2/service.rs
View file

@ -3,7 +3,7 @@ use std::marker::PhantomData;
use std::{io, net};
use actix_codec::{AsyncRead, AsyncWrite, Framed};
use actix_server_config::{Io, ServerConfig as SrvConfig};
use actix_server_config::{Io, IoStream, ServerConfig as SrvConfig};
use actix_service::{IntoNewService, NewService, Service};
use actix_utils::cloneable::CloneableService;
use bytes::Bytes;
@ -63,7 +63,7 @@ where
impl<T, P, S, B> NewService<SrvConfig> for H2Service<T, P, S, B>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: NewService<SrvConfig, Request = Request>,
S::Error: Into<Error>,
S::Response: Into<Response<B>>,
@ -95,7 +95,7 @@ pub struct H2ServiceResponse<T, P, S: NewService<SrvConfig, Request = Request>,
impl<T, P, S, B> Future for H2ServiceResponse<T, P, S, B>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: NewService<SrvConfig, Request = Request>,
S::Error: Into<Error>,
S::Response: Into<Response<B>>,
@ -140,7 +140,7 @@ where
impl<T, P, S, B> Service for H2ServiceHandler<T, P, S, B>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: Service<Request = Request>,
S::Error: Into<Error>,
S::Future: 'static,
@ -161,17 +161,20 @@ where
}
fn call(&mut self, req: Self::Request) -> Self::Future {
let io = req.into_parts().0;
let peer_addr = io.peer_addr();
H2ServiceHandlerResponse {
state: State::Handshake(
Some(self.srv.clone()),
Some(self.cfg.clone()),
server::handshake(req.into_parts().0),
peer_addr,
server::handshake(io),
),
}
}
}
enum State<T: AsyncRead + AsyncWrite, S: Service<Request = Request>, B: MessageBody>
enum State<T: IoStream, S: Service<Request = Request>, B: MessageBody>
where
S::Future: 'static,
{
@ -179,13 +182,14 @@ where
Handshake(
Option<CloneableService<S>>,
Option<ServiceConfig>,
Option<net::SocketAddr>,
Handshake<T, Bytes>,
),
}
pub struct H2ServiceHandlerResponse<T, S, B>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: Service<Request = Request>,
S::Error: Into<Error>,
S::Future: 'static,
@ -197,7 +201,7 @@ where
impl<T, S, B> Future for H2ServiceHandlerResponse<T, S, B>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: Service<Request = Request>,
S::Error: Into<Error>,
S::Future: 'static,
@ -210,24 +214,28 @@ where
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
match self.state {
State::Incoming(ref mut disp) => disp.poll(),
State::Handshake(ref mut srv, ref mut config, ref mut handshake) => {
match handshake.poll() {
Ok(Async::Ready(conn)) => {
self.state = State::Incoming(Dispatcher::new(
srv.take().unwrap(),
conn,
config.take().unwrap(),
None,
));
self.poll()
}
Ok(Async::NotReady) => Ok(Async::NotReady),
Err(err) => {
trace!("H2 handshake error: {}", err);
Err(err.into())
}
State::Handshake(
ref mut srv,
ref mut config,
ref peer_addr,
ref mut handshake,
) => match handshake.poll() {
Ok(Async::Ready(conn)) => {
self.state = State::Incoming(Dispatcher::new(
srv.take().unwrap(),
conn,
config.take().unwrap(),
None,
peer_addr.clone(),
));
self.poll()
}
}
Ok(Async::NotReady) => Ok(Async::NotReady),
Err(err) => {
trace!("H2 handshake error: {}", err);
Err(err.into())
}
},
}
}
}

3
actix-http/src/message.rs
View file

@ -1,4 +1,5 @@
use std::cell::{Ref, RefCell, RefMut};
use std::net;
use std::rc::Rc;
use bitflags::bitflags;
@ -43,6 +44,7 @@ pub struct RequestHead {
pub version: Version,
pub headers: HeaderMap,
pub extensions: RefCell<Extensions>,
pub peer_addr: Option<net::SocketAddr>,
flags: Flags,
}
@ -54,6 +56,7 @@ impl Default for RequestHead {
version: Version::HTTP_11,
headers: HeaderMap::with_capacity(16),
flags: Flags::empty(),
peer_addr: None,
extensions: RefCell::new(Extensions::new()),
}
}

12
actix-http/src/request.rs
View file

@ -1,5 +1,5 @@
use std::cell::{Ref, RefMut};
use std::fmt;
use std::{fmt, net};
use http::{header, Method, Uri, Version};
@ -139,6 +139,7 @@ impl<P> Request<P> {
}
/// Check if request requires connection upgrade
#[inline]
pub fn upgrade(&self) -> bool {
if let Some(conn) = self.head().headers.get(header::CONNECTION) {
if let Ok(s) = conn.to_str() {
@ -147,6 +148,15 @@ impl<P> Request<P> {
}
self.head().method == Method::CONNECT
}
/// Peer socket address
///
/// Peer address is actual socket address, if proxy is used in front of
/// actix http server, then peer address would be address of this proxy.
#[inline]
pub fn peer_addr(&self) -> Option<net::SocketAddr> {
self.head().peer_addr
}
}
impl<P> fmt::Debug for Request<P> {

64
actix-http/src/service.rs
View file

@ -1,8 +1,10 @@
use std::marker::PhantomData;
use std::{fmt, io};
use std::{fmt, io, net};
use actix_codec::{AsyncRead, AsyncWrite, Framed};
use actix_server_config::{Io as ServerIo, Protocol, ServerConfig as SrvConfig};
use actix_server_config::{
Io as ServerIo, IoStream, Protocol, ServerConfig as SrvConfig,
};
use actix_service::{IntoNewService, NewService, Service};
use actix_utils::cloneable::CloneableService;
use bytes::{Buf, BufMut, Bytes, BytesMut};
@ -128,7 +130,7 @@ where
impl<T, P, S, B, X, U> NewService<SrvConfig> for HttpService<T, P, S, B, X, U>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: NewService<SrvConfig, Request = Request>,
S::Error: Into<Error>,
S::InitError: fmt::Debug,
@ -182,7 +184,7 @@ pub struct HttpServiceResponse<
impl<T, P, S, B, X, U> Future for HttpServiceResponse<T, P, S, B, X, U>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: NewService<SrvConfig, Request = Request>,
S::Error: Into<Error>,
S::InitError: fmt::Debug,
@ -268,7 +270,7 @@ where
impl<T, P, S, B, X, U> Service for HttpServiceHandler<T, P, S, B, X, U>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: Service<Request = Request>,
S::Error: Into<Error>,
S::Future: 'static,
@ -317,6 +319,7 @@ where
let (io, _, proto) = req.into_parts();
match proto {
Protocol::Http2 => {
let peer_addr = io.peer_addr();
let io = Io {
inner: io,
unread: None,
@ -326,6 +329,7 @@ where
server::handshake(io),
self.cfg.clone(),
self.srv.clone(),
peer_addr,
))),
}
}
@ -357,7 +361,7 @@ where
S: Service<Request = Request>,
S::Future: 'static,
S::Error: Into<Error>,
T: AsyncRead + AsyncWrite,
T: IoStream,
B: MessageBody,
X: Service<Request = Request, Response = Request>,
X::Error: Into<Error>,
@ -376,12 +380,19 @@ where
Option<CloneableService<U>>,
)>,
),
Handshake(Option<(Handshake<Io<T>, Bytes>, ServiceConfig, CloneableService<S>)>),
Handshake(
Option<(
Handshake<Io<T>, Bytes>,
ServiceConfig,
CloneableService<S>,
Option<net::SocketAddr>,
)>,
),
}
pub struct HttpServiceHandlerResponse<T, S, B, X, U>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: Service<Request = Request>,
S::Error: Into<Error>,
S::Future: 'static,
@ -399,7 +410,7 @@ const HTTP2_PREFACE: [u8; 14] = *b"PRI * HTTP/2.0";
impl<T, S, B, X, U> Future for HttpServiceHandlerResponse<T, S, B, X, U>
where
T: AsyncRead + AsyncWrite,
T: IoStream,
S: Service<Request = Request>,
S::Error: Into<Error>,
S::Future: 'static,
@ -437,12 +448,17 @@ where
}
let (io, buf, cfg, srv, expect, upgrade) = data.take().unwrap();
if buf[..14] == HTTP2_PREFACE[..] {
let peer_addr = io.peer_addr();
let io = Io {
inner: io,
unread: Some(buf),
};
self.state =
State::Handshake(Some((server::handshake(io), cfg, srv)));
self.state = State::Handshake(Some((
server::handshake(io),
cfg,
srv,
peer_addr,
)));
} else {
self.state = State::H1(h1::Dispatcher::with_timeout(
io,
@ -470,8 +486,8 @@ where
} else {
panic!()
};
let (_, cfg, srv) = data.take().unwrap();
self.state = State::H2(Dispatcher::new(srv, conn, cfg, None));
let (_, cfg, srv, peer_addr) = data.take().unwrap();
self.state = State::H2(Dispatcher::new(srv, conn, cfg, None, peer_addr));
self.poll()
}
}
@ -523,3 +539,25 @@ impl<T: AsyncWrite> AsyncWrite for Io<T> {
self.inner.write_buf(buf)
}
}
impl<T: IoStream> IoStream for Io<T> {
#[inline]
fn peer_addr(&self) -> Option<net::SocketAddr> {
self.inner.peer_addr()
}
#[inline]
fn set_nodelay(&mut self, nodelay: bool) -> io::Result<()> {
self.inner.set_nodelay(nodelay)
}
#[inline]
fn set_linger(&mut self, dur: Option<std::time::Duration>) -> io::Result<()> {
self.inner.set_linger(dur)
}
#[inline]
fn set_keepalive(&mut self, dur: Option<std::time::Duration>) -> io::Result<()> {
self.inner.set_keepalive(dur)
}
}

15
actix-http/src/test.rs
View file

@ -4,6 +4,7 @@ use std::io;
use std::str::FromStr;
use actix_codec::{AsyncRead, AsyncWrite};
use actix_server_config::IoStream;
use bytes::{Buf, Bytes, BytesMut};
use futures::{Async, Poll};
use http::header::{self, HeaderName, HeaderValue};
@ -253,3 +254,17 @@ impl AsyncWrite for TestBuffer {
Ok(Async::NotReady)
}
}
impl IoStream for TestBuffer {
fn set_nodelay(&mut self, _nodelay: bool) -> io::Result<()> {
Ok(())
}
fn set_linger(&mut self, _dur: Option<std::time::Duration>) -> io::Result<()> {
Ok(())
}
fn set_keepalive(&mut self, _dur: Option<std::time::Duration>) -> io::Result<()> {
Ok(())
}
}

7
actix-http/tests/test_server.rs
View file

@ -35,7 +35,10 @@ fn test_h1() {
.keep_alive(KeepAlive::Disabled)
.client_timeout(1000)
.client_disconnect(1000)
.h1(|_| future::ok::<_, ()>(Response::Ok().finish()))
.h1(|req: Request| {
assert!(req.peer_addr().is_some());
future::ok::<_, ()>(Response::Ok().finish())
})
});
let response = srv.block_on(srv.get("/").send()).unwrap();
@ -50,6 +53,7 @@ fn test_h1_2() {
.client_timeout(1000)
.client_disconnect(1000)
.finish(|req: Request| {
assert!(req.peer_addr().is_some());
assert_eq!(req.version(), http::Version::HTTP_11);
future::ok::<_, ()>(Response::Ok().finish())
})
@ -115,6 +119,7 @@ fn test_h2_1() -> std::io::Result<()> {
.and_then(
HttpService::build()
.finish(|req: Request| {
assert!(req.peer_addr().is_some());
assert_eq!(req.version(), http::Version::HTTP_2);
future::ok::<_, Error>(Response::Ok().finish())
})