Skip to main content

sozu_command_lib/
request.rs

1use std::{
2    error,
3    fmt::{self, Display},
4    fs::File,
5    io::{BufReader, Read},
6    net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
7    str::FromStr,
8};
9
10use prost::{DecodeError, Message};
11use rusty_ulid::Ulid;
12
13use crate::{
14    proto::{
15        command::{
16            ClientAuthMode, ClientAuthPolicy, HttpsListenerConfig, InitialState, IpAddress,
17            LoadBalancingAlgorithms, PathRuleKind, Request, RequestHttpFrontend, RulePosition,
18            SocketAddress, Uint128, UpdateHttpsListenerConfig, WorkerRequest, ip_address,
19            request::RequestType,
20        },
21        display::format_request_type,
22    },
23    response::HttpFrontend,
24    state::ClusterId,
25};
26
27#[derive(thiserror::Error, Debug)]
28pub enum RequestError {
29    #[error("invalid value {value} for field '{name}'")]
30    InvalidValue { name: String, value: i32 },
31    #[error("Could not read requests from file: {0}")]
32    ReadFile(std::io::Error),
33    #[error("Could not decode requests: {0}")]
34    Decode(DecodeError),
35}
36
37impl Request {
38    /// determine to which of the four proxies (HTTP, HTTPS, TCP, UDP) a request is destined
39    pub fn get_destinations(&self) -> ProxyDestinations {
40        let mut proxy_destination = ProxyDestinations {
41            to_http_proxy: false,
42            to_https_proxy: false,
43            to_tcp_proxy: false,
44            to_udp_proxy: false,
45        };
46        let request_type = match &self.request_type {
47            Some(t) => t,
48            None => return proxy_destination,
49        };
50
51        match request_type {
52            RequestType::AddHttpFrontend(_) | RequestType::RemoveHttpFrontend(_) => {
53                proxy_destination.to_http_proxy = true
54            }
55
56            RequestType::AddHttpsFrontend(_)
57            | RequestType::RemoveHttpsFrontend(_)
58            | RequestType::AddCertificate(_)
59            | RequestType::QueryCertificatesFromWorkers(_)
60            | RequestType::ReplaceCertificate(_)
61            | RequestType::RemoveCertificate(_) => proxy_destination.to_https_proxy = true,
62
63            RequestType::AddTcpFrontend(_) | RequestType::RemoveTcpFrontend(_) => {
64                proxy_destination.to_tcp_proxy = true
65            }
66
67            RequestType::AddUdpFrontend(_) | RequestType::RemoveUdpFrontend(_) => {
68                proxy_destination.to_udp_proxy = true
69            }
70
71            RequestType::AddCluster(_)
72            | RequestType::AddBackend(_)
73            | RequestType::RemoveCluster(_)
74            | RequestType::RemoveBackend(_)
75            | RequestType::SetHealthCheck(_)
76            | RequestType::RemoveHealthCheck(_)
77            | RequestType::SoftStop(_)
78            | RequestType::HardStop(_)
79            | RequestType::Status(_) => {
80                proxy_destination.to_http_proxy = true;
81                proxy_destination.to_https_proxy = true;
82                proxy_destination.to_tcp_proxy = true;
83                proxy_destination.to_udp_proxy = true;
84            }
85
86            // handled at worker level prior to this call
87            RequestType::ConfigureMetrics(_)
88            | RequestType::SetMetricDetail(_)
89            | RequestType::QueryMetrics(_)
90            | RequestType::Logging(_)
91            | RequestType::QueryClustersHashes(_)
92            | RequestType::QueryClusterById(_)
93            | RequestType::QueryClustersByDomain(_)
94            | RequestType::SetMaxConnectionsPerIp(_)
95            | RequestType::QueryMaxConnectionsPerIp(_)
96            | RequestType::SetMaxConnectionsPerSubnet(_)
97            | RequestType::QueryMaxConnectionsPerSubnet(_)
98            | RequestType::SetMaxConnectionAttempts(_) => {}
99
100            // the Add***Listener / Update***Listener and other Listener orders will be
101            // handled separately by the notify_proxys function, so we don't give them
102            // destinations here
103            RequestType::AddHttpsListener(_)
104            | RequestType::AddHttpsListenerWithClientAuth(_)
105            | RequestType::AddHttpListener(_)
106            | RequestType::AddTcpListener(_)
107            | RequestType::AddUdpListener(_)
108            | RequestType::UpdateHttpListener(_)
109            | RequestType::UpdateHttpsListener(_)
110            | RequestType::UpdateHttpsListenerWithClientAuth(_)
111            | RequestType::UpdateTcpListener(_)
112            | RequestType::UpdateUdpListener(_)
113            | RequestType::RemoveListener(_)
114            | RequestType::ActivateListener(_)
115            | RequestType::DeactivateListener(_)
116            | RequestType::ReturnListenSockets(_) => {}
117
118            // These won't ever reach a worker anyway
119            RequestType::SaveState(_)
120            | RequestType::CountRequests(_)
121            | RequestType::QueryCertificatesFromTheState(_)
122            | RequestType::QueryHealthChecks(_)
123            | RequestType::LoadState(_)
124            | RequestType::ListWorkers(_)
125            | RequestType::ListFrontends(_)
126            | RequestType::ListListeners(_)
127            | RequestType::LaunchWorker(_)
128            | RequestType::UpgradeMain(_)
129            | RequestType::UpgradeWorker(_)
130            | RequestType::SubscribeEvents(_)
131            | RequestType::ReloadConfiguration(_) => {}
132        }
133
134        // POST: HTTP-frontend orders route to the HTTP proxy ONLY, HTTPS /
135        // certificate orders to the HTTPS proxy ONLY, and TCP-frontend orders
136        // to the TCP proxy ONLY — a frontend order must never fan out across
137        // protocol planes (that would double-apply the order). Cluster-wide
138        // and broadcast orders (AddCluster, SoftStop, …) legitimately target
139        // all three, so we only assert the single-plane exclusivity here.
140        debug_assert!(
141            !(proxy_destination.to_http_proxy
142                && proxy_destination.to_https_proxy
143                && proxy_destination.to_tcp_proxy)
144                || matches!(
145                    self.request_type,
146                    Some(
147                        RequestType::AddCluster(_)
148                            | RequestType::AddBackend(_)
149                            | RequestType::RemoveCluster(_)
150                            | RequestType::RemoveBackend(_)
151                            | RequestType::SetHealthCheck(_)
152                            | RequestType::RemoveHealthCheck(_)
153                            | RequestType::SoftStop(_)
154                            | RequestType::HardStop(_)
155                            | RequestType::Status(_)
156                    )
157                ),
158            "only cluster-wide / broadcast orders may target all three proxy planes"
159        );
160        // POST: a None request_type carries no destination at all.
161        debug_assert!(
162            self.request_type.is_some()
163                || (!proxy_destination.to_http_proxy
164                    && !proxy_destination.to_https_proxy
165                    && !proxy_destination.to_tcp_proxy),
166            "a request without a request_type must have no proxy destination"
167        );
168        proxy_destination
169    }
170
171    /// True if the request is a SoftStop or a HardStop
172    pub fn is_a_stop(&self) -> bool {
173        matches!(
174            self.request_type,
175            Some(RequestType::SoftStop(_)) | Some(RequestType::HardStop(_))
176        )
177    }
178
179    pub fn short_name(&self) -> &str {
180        match &self.request_type {
181            Some(request_type) => format_request_type(request_type),
182            None => "Unallowed",
183        }
184    }
185}
186
187impl WorkerRequest {
188    pub fn new(id: String, content: Request) -> Self {
189        Self { id, content }
190    }
191}
192
193impl HttpsListenerConfig {
194    /// True when the listener asks for a client certificate.
195    ///
196    /// Any value other than `CLIENT_AUTH_NONE` counts, including one this
197    /// build does not know: it must travel on the verb a pre-mTLS peer
198    /// cannot decode, and the worker then rejects it when it builds the
199    /// listener.
200    pub fn requests_client_auth(&self) -> bool {
201        self.client_auth
202            .is_some_and(|mode| mode != ClientAuthMode::ClientAuthNone as i32)
203    }
204
205    /// The listener's client authentication policy, as a patch would carry it.
206    pub fn client_auth_policy(&self) -> ClientAuthPolicy {
207        ClientAuthPolicy {
208            client_auth: self.client_auth,
209            client_ca_certificates: self.client_ca_certificates.clone(),
210            client_ca_crls: self.client_ca_crls.clone(),
211        }
212    }
213
214    /// Replace the client authentication policy with `policy`, all three
215    /// fields at once: a CA or CRL the policy leaves out is dropped. The main
216    /// process and the workers both apply a patch through this, so their
217    /// copies of the listener cannot diverge.
218    pub fn apply_client_auth_policy(&mut self, policy: &ClientAuthPolicy) {
219        self.client_auth = policy.client_auth;
220        self.client_ca_certificates = policy.client_ca_certificates.clone();
221        self.client_ca_crls = policy.client_ca_crls.clone();
222    }
223}
224
225impl RequestType {
226    /// The request that adds this HTTPS listener, on the verb its client
227    /// authentication policy calls for.
228    ///
229    /// A listener that asks for a client certificate goes out as
230    /// `AddHttpsListenerWithClientAuth`, which a worker or main process that
231    /// predates mutual TLS cannot decode, so it fails the request instead of
232    /// building the listener without client authentication. Any other
233    /// listener keeps `AddHttpsListener`, which every version understands.
234    pub fn add_https_listener(listener: HttpsListenerConfig) -> Self {
235        if listener.requests_client_auth() {
236            RequestType::AddHttpsListenerWithClientAuth(listener)
237        } else {
238            RequestType::AddHttpsListener(listener)
239        }
240    }
241
242    /// The request that applies this HTTPS listener patch: a patch that
243    /// replaces the client authentication policy goes out as
244    /// `UpdateHttpsListenerWithClientAuth`, which a worker that predates the
245    /// policy field cannot decode, so it fails the patch instead of keeping
246    /// the CA and CRLs it had. Any other patch keeps `UpdateHttpsListener`.
247    pub fn update_https_listener(patch: UpdateHttpsListenerConfig) -> Self {
248        if patch.client_auth_policy.is_some() {
249            RequestType::UpdateHttpsListenerWithClientAuth(patch)
250        } else {
251            RequestType::UpdateHttpsListener(patch)
252        }
253    }
254
255    /// The same request, with an HTTPS listener addition or patch moved to
256    /// the verb [`RequestType::add_https_listener`] or
257    /// [`RequestType::update_https_listener`] picks for it. Every other
258    /// request is returned unchanged.
259    ///
260    /// The main process applies this to every request it fans out to the
261    /// workers, so a client that sends client authentication on a historical
262    /// verb cannot get it silently dropped by a worker that predates it.
263    pub fn into_canonical(self) -> Self {
264        match self {
265            RequestType::AddHttpsListener(listener)
266            | RequestType::AddHttpsListenerWithClientAuth(listener) => {
267                RequestType::add_https_listener(listener)
268            }
269            RequestType::UpdateHttpsListener(patch)
270            | RequestType::UpdateHttpsListenerWithClientAuth(patch) => {
271                RequestType::update_https_listener(patch)
272            }
273            other => other,
274        }
275    }
276}
277
278impl Request {
279    /// [`RequestType::into_canonical`] applied to the request type.
280    pub fn into_canonical(self) -> Self {
281        Request {
282            request_type: self.request_type.map(RequestType::into_canonical),
283        }
284    }
285}
286
287impl fmt::Display for WorkerRequest {
288    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
289        write!(f, "{}-{:?}", self.id, self.content)
290    }
291}
292
293pub fn read_initial_state_from_file(file: &mut File) -> Result<InitialState, RequestError> {
294    let mut buf_reader = BufReader::new(file);
295    read_initial_state(&mut buf_reader)
296}
297
298pub fn read_initial_state<R: Read>(reader: &mut R) -> Result<InitialState, RequestError> {
299    let mut buffer = Vec::new();
300    reader
301        .read_to_end(&mut buffer)
302        .map_err(RequestError::ReadFile)?;
303
304    InitialState::decode(&buffer[..]).map_err(RequestError::Decode)
305}
306
307#[derive(Debug, Clone, PartialEq, Eq, Hash)]
308pub struct ProxyDestinations {
309    pub to_http_proxy: bool,
310    pub to_https_proxy: bool,
311    pub to_tcp_proxy: bool,
312    pub to_udp_proxy: bool,
313}
314
315impl RequestHttpFrontend {
316    /// convert a requested frontend to a usable one by parsing its address
317    pub fn to_frontend(self) -> Result<HttpFrontend, RequestError> {
318        let requested_hostname = self.hostname.clone();
319        let requested_cluster_id = self.cluster_id.clone();
320        let frontend = HttpFrontend {
321            address: self.address.into(),
322            cluster_id: self.cluster_id.map(ClusterId::from),
323            hostname: self.hostname,
324            path: self.path,
325            method: self.method,
326            position: RulePosition::try_from(self.position).map_err(|_| {
327                RequestError::InvalidValue {
328                    name: "position".to_string(),
329                    value: self.position,
330                }
331            })?,
332            tags: Some(self.tags),
333            redirect: self.redirect,
334            redirect_scheme: self.redirect_scheme,
335            redirect_template: self.redirect_template,
336            rewrite_host: self.rewrite_host,
337            rewrite_path: self.rewrite_path,
338            rewrite_port: self.rewrite_port,
339            required_auth: self.required_auth,
340            headers: self.headers,
341            hsts: self.hsts,
342        };
343
344        // POST: routing identity (hostname + cluster_id) is carried through
345        // unchanged — only the address is reparsed and the position is mapped
346        // through the proto enum. A frontend whose hostname or cluster shifted
347        // here would route traffic to the wrong place.
348        debug_assert_eq!(
349            frontend.hostname, requested_hostname,
350            "hostname must survive the frontend conversion"
351        );
352        debug_assert_eq!(
353            frontend.cluster_id.as_deref(),
354            requested_cluster_id.as_deref(),
355            "cluster_id must survive the frontend conversion"
356        );
357        Ok(frontend)
358    }
359}
360
361impl Display for RequestHttpFrontend {
362    /// Used to create a unique summary of the frontend, used as a key in maps
363    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
364        let s = match &PathRuleKind::try_from(self.path.kind) {
365            Ok(PathRuleKind::Prefix) => {
366                format!("{};{};P{}", self.address, self.hostname, self.path.value)
367            }
368            Ok(PathRuleKind::Regex) => {
369                format!("{};{};R{}", self.address, self.hostname, self.path.value)
370            }
371            Ok(PathRuleKind::Equals) => {
372                format!("{};{};={}", self.address, self.hostname, self.path.value)
373            }
374            Err(e) => format!("Wrong variant of PathRuleKind: {e}"),
375        };
376
377        match &self.method {
378            Some(method) => write!(f, "{s};{method}"),
379            None => write!(f, "{s}"),
380        }
381    }
382}
383
384#[derive(Debug)]
385pub struct ParseErrorLoadBalancing;
386
387impl fmt::Display for ParseErrorLoadBalancing {
388    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
389        write!(f, "Cannot find the load balancing policy asked")
390    }
391}
392
393impl error::Error for ParseErrorLoadBalancing {
394    fn description(&self) -> &str {
395        "Cannot find the load balancing policy asked"
396    }
397
398    fn cause(&self) -> Option<&dyn error::Error> {
399        None
400    }
401}
402
403impl FromStr for LoadBalancingAlgorithms {
404    type Err = ParseErrorLoadBalancing;
405
406    fn from_str(s: &str) -> Result<Self, Self::Err> {
407        match s.to_lowercase().as_str() {
408            "round_robin" => Ok(LoadBalancingAlgorithms::RoundRobin),
409            "random" => Ok(LoadBalancingAlgorithms::Random),
410            "power_of_two" => Ok(LoadBalancingAlgorithms::PowerOfTwo),
411            "least_loaded" => Ok(LoadBalancingAlgorithms::LeastLoaded),
412            "hrw" => Ok(LoadBalancingAlgorithms::Hrw),
413            "maglev" => Ok(LoadBalancingAlgorithms::Maglev),
414            _ => Err(ParseErrorLoadBalancing {}),
415        }
416    }
417}
418
419impl SocketAddress {
420    pub fn new_v4(a: u8, b: u8, c: u8, d: u8, port: u16) -> Self {
421        SocketAddr::new(IpAddr::V4(Ipv4Addr::new(a, b, c, d)), port).into()
422    }
423}
424
425impl From<SocketAddr> for SocketAddress {
426    fn from(socket_addr: SocketAddr) -> SocketAddress {
427        let ip_inner = match socket_addr {
428            SocketAddr::V4(ip_v4_addr) => ip_address::Inner::V4(u32::from(*ip_v4_addr.ip())),
429            SocketAddr::V6(ip_v6_addr) => {
430                ip_address::Inner::V6(Uint128::from(u128::from(*ip_v6_addr.ip())))
431            }
432        };
433
434        let encoded = SocketAddress {
435            port: socket_addr.port() as u32,
436            ip: IpAddress {
437                inner: Some(ip_inner),
438            },
439        };
440
441        // POST: the port widens losslessly (u16 → u32) and the proto address
442        // family matches the source family — a V4 SocketAddr must never encode
443        // as a V6 inner and vice versa, or the reverse `From` would synthesize
444        // the wrong address.
445        debug_assert_eq!(
446            encoded.port,
447            socket_addr.port() as u32,
448            "port must round-trip losslessly into the proto"
449        );
450        debug_assert_eq!(
451            matches!(encoded.ip.inner, Some(ip_address::Inner::V4(_))),
452            socket_addr.is_ipv4(),
453            "proto IP family must match the source SocketAddr family"
454        );
455        encoded
456    }
457}
458
459impl From<SocketAddress> for SocketAddr {
460    fn from(socket_address: SocketAddress) -> Self {
461        // PRE: a wire-sourced proto port may exceed u16::MAX (16-bit on the
462        // wire is carried as a 32-bit field). This is peer/config input, so we
463        // narrow rather than panic; the debug_assert only guards our *own*
464        // encoders, which never emit an out-of-range port.
465        debug_assert!(
466            socket_address.port <= u16::MAX as u32,
467            "self-encoded proto port must fit in a u16"
468        );
469        let had_inner = socket_address.ip.inner.is_some();
470        let port = socket_address.port as u16;
471
472        let ip = match socket_address.ip.inner {
473            Some(inner) => match inner {
474                ip_address::Inner::V4(v4_value) => IpAddr::V4(Ipv4Addr::from(v4_value)),
475                ip_address::Inner::V6(v6_value) => IpAddr::V6(Ipv6Addr::from(u128::from(v6_value))),
476            },
477            None => IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), // should never happen
478        };
479
480        let decoded = SocketAddr::new(ip, port);
481        // POST: a self-encoded proto address always carries an inner IP, so the
482        // unspecified-V4 fallback is only ever reached on malformed peer input.
483        debug_assert!(
484            had_inner || decoded.ip() == IpAddr::V4(Ipv4Addr::UNSPECIFIED),
485            "missing inner IP must decode to the unspecified-V4 sentinel"
486        );
487        decoded
488    }
489}
490
491impl From<Uint128> for u128 {
492    fn from(value: Uint128) -> Self {
493        let combined = value.low as u128 | ((value.high as u128) << 64);
494        // POST: the two 64-bit halves occupy disjoint bit ranges, so the low
495        // half is recoverable as the bottom 64 bits and the high half as the
496        // top 64 bits — the pack is bijective.
497        debug_assert_eq!(
498            combined as u64, value.low,
499            "low half must be the bottom 64 bits"
500        );
501        debug_assert_eq!(
502            (combined >> 64) as u64,
503            value.high,
504            "high half must be the top 64 bits"
505        );
506        combined
507    }
508}
509
510impl From<u128> for Uint128 {
511    fn from(value: u128) -> Self {
512        let low = value as u64;
513        let high = (value >> 64) as u64;
514        let packed = Uint128 { low, high };
515        // POST: splitting then recombining reproduces the original u128 — the
516        // split-into-halves and join-from-halves operations are mutual
517        // inverses (no bit is lost or duplicated).
518        debug_assert_eq!(
519            u128::from(Uint128 {
520                low: packed.low,
521                high: packed.high
522            }),
523            value,
524            "u128 → Uint128 → u128 must round-trip"
525        );
526        packed
527    }
528}
529
530impl From<i128> for Uint128 {
531    fn from(value: i128) -> Self {
532        Uint128::from(value as u128)
533    }
534}
535
536impl From<Ulid> for Uint128 {
537    fn from(value: Ulid) -> Self {
538        let (low, high) = value.into();
539        let packed = Uint128 { low, high };
540        // POST: the (low, high) tuple is carried verbatim into the proto, so
541        // re-reading it reconstructs the same Ulid — the encoding loses no bits.
542        debug_assert_eq!(
543            Ulid::from((packed.low, packed.high)),
544            value,
545            "Ulid → Uint128 must preserve all 128 bits"
546        );
547        packed
548    }
549}
550
551impl From<Uint128> for Ulid {
552    fn from(value: Uint128) -> Self {
553        let Uint128 { low, high } = value;
554        let ulid = Ulid::from((low, high));
555        // POST: the decode is the exact inverse of the encode above — the same
556        // halves go back out, so Uint128 → Ulid → Uint128 round-trips.
557        debug_assert_eq!(
558            Uint128::from(ulid),
559            value,
560            "Uint128 → Ulid must preserve all 128 bits"
561        );
562        ulid
563    }
564}
565
566#[cfg(test)]
567mod tests {
568    use prost::Message;
569
570    use super::*;
571
572    fn https_listener(client_auth: Option<i32>) -> HttpsListenerConfig {
573        HttpsListenerConfig {
574            address: SocketAddress::new_v4(127, 0, 0, 1, 8443),
575            client_auth,
576            client_ca_certificates: vec!["CLIENT CA PEM".to_owned()],
577            ..Default::default()
578        }
579    }
580
581    const NONE: i32 = ClientAuthMode::ClientAuthNone as i32;
582    const OPTIONAL: i32 = ClientAuthMode::ClientAuthOptional as i32;
583    const REQUIRED: i32 = ClientAuthMode::ClientAuthRequired as i32;
584
585    #[test]
586    fn add_https_listener_picks_the_verb_from_the_client_auth_mode() {
587        for client_auth in [None, Some(NONE)] {
588            assert!(
589                matches!(
590                    RequestType::add_https_listener(https_listener(client_auth)),
591                    RequestType::AddHttpsListener(_)
592                ),
593                "a listener without client auth ({client_auth:?}) keeps the verb every worker decodes"
594            );
595        }
596        // 999 names no mode: it must not travel on the verb an old worker
597        // would decode and silently strip, the worker rejects it at build.
598        for client_auth in [Some(OPTIONAL), Some(REQUIRED), Some(999)] {
599            assert!(
600                matches!(
601                    RequestType::add_https_listener(https_listener(client_auth)),
602                    RequestType::AddHttpsListenerWithClientAuth(_)
603                ),
604                "a listener asking for a client certificate ({client_auth:?}) needs the mTLS verb"
605            );
606        }
607    }
608
609    #[test]
610    fn into_canonical_moves_https_listeners_to_their_verb_both_ways() {
611        let required = https_listener(Some(REQUIRED));
612        assert_eq!(
613            RequestType::AddHttpsListener(required.clone()).into_canonical(),
614            RequestType::AddHttpsListenerWithClientAuth(required.clone()),
615            "client auth sent on the historical verb must be moved to the mTLS verb"
616        );
617        assert_eq!(
618            RequestType::AddHttpsListenerWithClientAuth(required.clone()).into_canonical(),
619            RequestType::AddHttpsListenerWithClientAuth(required),
620        );
621
622        let none = https_listener(None);
623        assert_eq!(
624            RequestType::AddHttpsListenerWithClientAuth(none.clone()).into_canonical(),
625            RequestType::AddHttpsListener(none.clone()),
626            "a listener without client auth goes back to the verb every worker decodes"
627        );
628        assert_eq!(
629            Request::from(RequestType::AddHttpsListener(none.clone())).into_canonical(),
630            Request::from(RequestType::AddHttpsListener(none)),
631        );
632
633        let other = RequestType::SoftStop(Default::default());
634        assert_eq!(other.clone().into_canonical(), other);
635        assert_eq!(
636            Request { request_type: None }.into_canonical(),
637            Request { request_type: None }
638        );
639    }
640
641    /// The `WorkerRequest` schema of a worker that predates mutual TLS, cut
642    /// down to what matters here: the envelope, and the one verb such a
643    /// worker had for adding an HTTPS listener. prost skips every tag a
644    /// schema does not declare, exactly as it does for the verbs elided here.
645    #[derive(Clone, PartialEq, Message)]
646    struct LegacyWorkerRequest {
647        #[prost(string, tag = "1")]
648        id: String,
649        #[prost(message, optional, tag = "2")]
650        content: Option<LegacyRequest>,
651    }
652
653    #[derive(Clone, PartialEq, Message)]
654    struct LegacyRequest {
655        #[prost(oneof = "LegacyRequestType", tags = "27, 48")]
656        request_type: Option<LegacyRequestType>,
657    }
658
659    #[derive(Clone, PartialEq, prost::Oneof)]
660    enum LegacyRequestType {
661        /// `add_https_listener`, holding an `HttpsListenerConfig` that has no
662        /// client authentication field.
663        #[prost(message, tag = "27")]
664        AddHttpsListener(LegacyHttpsListenerConfig),
665        /// `update_https_listener`, holding an `UpdateHttpsListenerConfig`
666        /// that has no client authentication policy field.
667        #[prost(message, tag = "48")]
668        UpdateHttpsListener(LegacyHttpsListenerConfig),
669    }
670
671    #[derive(Clone, PartialEq, Message)]
672    struct LegacyHttpsListenerConfig {
673        #[prost(message, optional, tag = "1")]
674        address: Option<SocketAddress>,
675    }
676
677    fn decode_on_a_legacy_worker(request: RequestType) -> Option<LegacyRequestType> {
678        let bytes = WorkerRequest::new("ID-1".to_owned(), request.into()).encode_to_vec();
679        let legacy = LegacyWorkerRequest::decode(bytes.as_slice())
680            .expect("a legacy worker decodes the envelope of every request");
681        assert_eq!(legacy.id, "ID-1");
682        legacy
683            .content
684            .expect("the envelope always carries a request")
685            .request_type
686    }
687
688    /// A main process that knows mutual TLS talking to a worker that does
689    /// not, the state between `UpgradeMain` and the replacement of the last
690    /// old worker.
691    #[test]
692    fn a_worker_that_predates_mtls_cannot_decode_a_listener_that_requires_it() {
693        let required = https_listener(Some(REQUIRED));
694
695        // The failure mode the dedicated verb exists for: on the historical
696        // verb, the old worker sees a plain HTTPS listener, client auth gone.
697        assert!(
698            matches!(
699                decode_on_a_legacy_worker(RequestType::AddHttpsListener(required.clone())),
700                Some(LegacyRequestType::AddHttpsListener(_))
701            ),
702            "precondition: the historical verb decodes on an old worker without its client auth"
703        );
704
705        // What the main process actually sends: nothing the old worker can
706        // build a listener from, so it fails the request instead.
707        assert_eq!(
708            decode_on_a_legacy_worker(
709                RequestType::AddHttpsListener(required.clone()).into_canonical()
710            ),
711            None,
712            "a listener that requires a client certificate must be undecodable by an old worker"
713        );
714        assert_eq!(
715            decode_on_a_legacy_worker(RequestType::add_https_listener(required)),
716            None
717        );
718
719        // Positive space: a listener without client auth stays readable, so
720        // a mixed fleet keeps serving every listener it can serve safely.
721        assert_eq!(
722            decode_on_a_legacy_worker(RequestType::add_https_listener(https_listener(None))),
723            Some(LegacyRequestType::AddHttpsListener(
724                LegacyHttpsListenerConfig {
725                    address: Some(SocketAddress::new_v4(127, 0, 0, 1, 8443)),
726                }
727            ))
728        );
729    }
730
731    fn policy_patch(policy: Option<ClientAuthPolicy>) -> UpdateHttpsListenerConfig {
732        UpdateHttpsListenerConfig {
733            address: SocketAddress::new_v4(127, 0, 0, 1, 8443),
734            client_auth_policy: policy,
735            ..Default::default()
736        }
737    }
738
739    #[test]
740    fn update_https_listener_picks_the_verb_from_the_policy_field() {
741        assert!(matches!(
742            RequestType::update_https_listener(policy_patch(None)),
743            RequestType::UpdateHttpsListener(_)
744        ));
745        // Any policy, even one turning client auth off: a worker that ignored
746        // it would keep enforcing, or keep trusting, what the patch replaces.
747        for client_auth in [None, Some(NONE), Some(REQUIRED)] {
748            let policy = ClientAuthPolicy {
749                client_auth,
750                ..Default::default()
751            };
752            assert!(
753                matches!(
754                    RequestType::update_https_listener(policy_patch(Some(policy))),
755                    RequestType::UpdateHttpsListenerWithClientAuth(_)
756                ),
757                "a patch carrying a policy ({client_auth:?}) needs the mTLS verb"
758            );
759        }
760        let with_policy = policy_patch(Some(ClientAuthPolicy::default()));
761        assert_eq!(
762            RequestType::UpdateHttpsListener(with_policy.clone()).into_canonical(),
763            RequestType::UpdateHttpsListenerWithClientAuth(with_policy)
764        );
765        assert_eq!(
766            RequestType::UpdateHttpsListenerWithClientAuth(policy_patch(None)).into_canonical(),
767            RequestType::UpdateHttpsListener(policy_patch(None))
768        );
769    }
770
771    #[test]
772    fn a_worker_that_predates_mtls_cannot_decode_a_client_auth_policy_patch() {
773        let refresh = policy_patch(Some(https_listener(Some(REQUIRED)).client_auth_policy()));
774        assert!(
775            matches!(
776                decode_on_a_legacy_worker(RequestType::UpdateHttpsListener(refresh.clone())),
777                Some(LegacyRequestType::UpdateHttpsListener(_))
778            ),
779            "precondition: the historical verb decodes on an old worker without its policy"
780        );
781        assert_eq!(
782            decode_on_a_legacy_worker(RequestType::update_https_listener(refresh)),
783            None,
784            "a patch replacing the client auth policy must be undecodable by an old worker"
785        );
786        assert!(matches!(
787            decode_on_a_legacy_worker(RequestType::update_https_listener(policy_patch(None))),
788            Some(LegacyRequestType::UpdateHttpsListener(_))
789        ));
790    }
791
792    #[test]
793    fn applying_a_client_auth_policy_replaces_all_of_it() {
794        let mut listener = https_listener(Some(REQUIRED));
795        listener.client_ca_crls = vec!["OLD CRL".to_owned()];
796        let policy = ClientAuthPolicy {
797            client_auth: Some(OPTIONAL),
798            client_ca_certificates: vec!["NEW CA".to_owned()],
799            client_ca_crls: vec![],
800        };
801        listener.apply_client_auth_policy(&policy);
802        assert_eq!(listener.client_auth_policy(), policy);
803        assert!(
804            listener.client_ca_crls.is_empty(),
805            "a CRL the policy leaves out is dropped, not kept"
806        );
807    }
808}