cronoq-server 0.2.1

gRPC/HTTP server for the cronoq scheduler with remote workers
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

//! gRPC-Web on the gRPC listener (opt-in with `[server.grpc_web]`).
//!
//! Browsers speak gRPC-Web over HTTP/1.1 or HTTP/2. Requests pass, outermost first:
//! - CORS, answering preflights for the configured origins only;
//! - a guard that, without credentials, answers only requests for a loopback host (as the HTTP
//!   API does, against DNS rebinding) and refuses HTTP/1.1 requests that are not gRPC-Web;
//! - tonic-web, which translates gRPC-Web to gRPC for the services, where authentication and
//!   authorization run as for any other call.

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;

use ::http::{HeaderName, HeaderValue, Method, Request, Response, Version, header};
use tower::{Layer, Service, ServiceBuilder};
use tower_http::cors::{AllowOrigin, CorsLayer};

use crate::config::GrpcWebConfig;

/// The middleware stack for a gRPC listener with gRPC-Web enabled.
pub type GrpcWebStack = ServiceBuilder<
    tower::layer::util::Stack<
        tonic_web::GrpcWebLayer,
        tower::layer::util::Stack<
            GuardLayer,
            tower::layer::util::Stack<CorsLayer, tower::layer::util::Identity>,
        >,
    >,
>;

/// Build the stack. `loopback_only` is set when the server runs without credentials.
pub fn layers(cfg: &GrpcWebConfig, loopback_only: bool) -> GrpcWebStack {
    let origins: Vec<HeaderValue> = cfg
        .allowed_origins
        .iter()
        // Browsers send origins in lower case; schemes and hosts are case-insensitive.
        .map(|o| {
            HeaderValue::from_str(&o.to_ascii_lowercase())
                .expect("origins are validated with the configuration")
        })
        .collect();
    let cors = CorsLayer::new()
        .allow_origin(AllowOrigin::list(origins))
        .allow_methods([Method::POST])
        .allow_headers([
            header::AUTHORIZATION,
            header::CONTENT_TYPE,
            HeaderName::from_static("x-grpc-web"),
            HeaderName::from_static("x-user-agent"),
            HeaderName::from_static("grpc-timeout"),
        ])
        .expose_headers([
            HeaderName::from_static("grpc-status"),
            HeaderName::from_static("grpc-message"),
            HeaderName::from_static("grpc-status-details-bin"),
        ])
        .max_age(Duration::from_secs(600));
    ServiceBuilder::new()
        .layer(cors)
        .layer(GuardLayer { loopback_only })
        .layer(tonic_web::GrpcWebLayer::new())
}

/// See the module documentation.
#[derive(Clone, Copy, Debug)]
pub struct GuardLayer {
    loopback_only: bool,
}

impl<S> Layer<S> for GuardLayer {
    type Service = Guard<S>;
    fn layer(&self, inner: S) -> Guard<S> {
        Guard {
            inner,
            loopback_only: self.loopback_only,
        }
    }
}

#[derive(Clone, Debug)]
pub struct Guard<S> {
    inner: S,
    loopback_only: bool,
}

fn is_grpc_web<B>(req: &Request<B>) -> bool {
    req.headers()
        .get(header::CONTENT_TYPE)
        .and_then(|v| v.to_str().ok())
        .is_some_and(|ct| ct.starts_with("application/grpc-web"))
}

impl<S, B, RB> Service<Request<B>> for Guard<S>
where
    S: Service<Request<B>, Response = Response<RB>>,
    S::Future: Send + 'static,
    RB: Default,
{
    type Response = Response<RB>;
    type Error = S::Error;
    type Future = Pin<Box<dyn Future<Output = Result<Response<RB>, S::Error>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), S::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, req: Request<B>) -> Self::Future {
        let refuse = if self.loopback_only && !crate::http::loopback_host(&req) {
            Some(tonic::Status::permission_denied(
                "without auth the server only answers requests for a loopback host",
            ))
        } else if req.version() < Version::HTTP_2 && !is_grpc_web(&req) {
            Some(tonic::Status::unimplemented(
                "HTTP/1.1 carries only gRPC-Web requests; use HTTP/2 for gRPC",
            ))
        } else {
            None
        };
        match refuse {
            Some(status) => Box::pin(async move { Ok(status.into_http()) }),
            None => Box::pin(self.inner.call(req)),
        }
    }
}