rocketmq_auth/
remoting_auth_context.rs

1// Copyright 2026 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::net::SocketAddr;
16
17use rocketmq_transport::api::EmbeddedCaller;
18use rocketmq_transport::api::RemotingRequest;
19use rocketmq_transport::api::RequestOrigin;
20
21use crate::AuthFailureKind;
22use crate::AuthOperation;
23use crate::AuthServiceError;
24use crate::AuthServiceResult;
25use rocketmq_transport::api::SessionView;
26
27/// Typed, read-only transport facts used by remoting authentication and authorization.
28///
29/// This value carries no connection, writer, cancellation, or lifecycle authority.
30#[derive(Clone, Debug, Default, Eq, PartialEq)]
31pub struct RemotingAuthContext {
32    source_ip: Option<String>,
33    channel_id: Option<String>,
34    embedded: bool,
35}
36
37impl RemotingAuthContext {
38    /// Creates a compatibility authentication context from ingress facts.
39    ///
40    /// New network adapters should use [`Self::network`]. Embedded trust can
41    /// only be projected by [`Self::from_request`] from a transport-owned
42    /// request; missing network metadata never implies embedded trust.
43    #[must_use]
44    pub fn new(source_ip: Option<String>, channel_id: Option<String>) -> Self {
45        Self {
46            source_ip,
47            channel_id,
48            embedded: false,
49        }
50    }
51
52    /// Creates a context for a network request from trusted ingress metadata.
53    #[must_use]
54    pub fn network(source_ip: impl Into<String>, channel_id: impl Into<String>) -> Self {
55        Self {
56            source_ip: Some(source_ip.into()),
57            channel_id: Some(channel_id.into()),
58            embedded: false,
59        }
60    }
61
62    /// Creates a context for a trusted in-process request.
63    #[must_use]
64    pub(crate) fn embedded(channel_id: impl Into<String>) -> Self {
65        Self {
66            source_ip: None,
67            channel_id: Some(channel_id.into()),
68            embedded: true,
69        }
70    }
71
72    /// Projects authentication facts from a request assembled by the trusted
73    /// transport boundary.
74    ///
75    /// # Errors
76    ///
77    /// Returns an authentication error if origin and session kinds disagree or
78    /// required network/session metadata is missing.
79    pub fn from_request(request: &RemotingRequest) -> AuthServiceResult<Self> {
80        let channel_id = format!(
81            "transport-session-{}",
82            request.original_identity().request_id().owner_id()
83        );
84        let context = match (request.origin(), request.session()) {
85            (RequestOrigin::Network { peer }, SessionView::Network { remote_addr, .. }) => {
86                validate_network_origin(peer.address(), *remote_addr)?;
87                Self::network(remote_addr.ip().to_string(), channel_id)
88            }
89            (RequestOrigin::Embedded { caller }, SessionView::Embedded { .. }) => {
90                validate_embedded_caller(*caller)?;
91                Self::embedded(channel_id)
92            }
93            _ => {
94                return Err(invalid_remoting_context());
95            }
96        };
97        context.validate()?;
98        Ok(context)
99    }
100
101    /// Returns the trusted source IP, when the ingress has a network peer.
102    #[must_use]
103    pub fn source_ip(&self) -> Option<&str> {
104        self.source_ip.as_deref()
105    }
106
107    /// Returns the stable transport session identity used for request signing.
108    #[must_use]
109    pub fn channel_id(&self) -> Option<&str> {
110        self.channel_id.as_deref()
111    }
112
113    /// Returns whether the request came from a trusted in-process adapter.
114    #[must_use]
115    pub const fn is_embedded(&self) -> bool {
116        self.embedded
117    }
118
119    /// Verifies that the context carries a complete, internally consistent
120    /// trusted-ingress projection.
121    ///
122    /// # Errors
123    ///
124    /// Returns an authentication error for missing session identity, missing
125    /// network source, or mixed embedded/network facts.
126    pub fn validate(&self) -> AuthServiceResult<()> {
127        let channel_id = self.channel_id().filter(|value| !value.trim().is_empty());
128        if channel_id.is_none() {
129            return Err(invalid_remoting_context());
130        }
131        if self.embedded {
132            if self.source_ip.is_some() {
133                return Err(invalid_remoting_context());
134            }
135            return Ok(());
136        }
137        if self.source_ip().filter(|value| !value.trim().is_empty()).is_none() {
138            return Err(invalid_remoting_context());
139        }
140        Ok(())
141    }
142}
143
144fn validate_network_origin(peer: SocketAddr, remote_addr: SocketAddr) -> AuthServiceResult<()> {
145    if peer != remote_addr {
146        return Err(invalid_remoting_context());
147    }
148    Ok(())
149}
150
151fn validate_embedded_caller(caller: EmbeddedCaller) -> AuthServiceResult<()> {
152    match caller {
153        EmbeddedCaller::BrokerProxy => Ok(()),
154        _ => Err(invalid_remoting_context()),
155    }
156}
157
158fn invalid_remoting_context() -> AuthServiceError {
159    AuthServiceError::new(AuthOperation::BuildContext, AuthFailureKind::Unauthenticated)
160}
161
162#[cfg(test)]
163mod tests {
164    use std::net::SocketAddr;
165
166    use rocketmq_transport::api::EmbeddedCaller;
167
168    use super::validate_embedded_caller;
169    use super::validate_network_origin;
170    use super::RemotingAuthContext;
171
172    #[test]
173    fn exposes_only_owned_authentication_facts() {
174        let context = RemotingAuthContext::network("192.0.2.10", "transport-session-17");
175
176        assert_eq!(context.source_ip(), Some("192.0.2.10"));
177        assert_eq!(context.channel_id(), Some("transport-session-17"));
178    }
179
180    #[test]
181    fn embedded_context_can_omit_network_facts() {
182        let context = RemotingAuthContext::embedded("embedded-session-1");
183
184        assert_eq!(context.source_ip(), None);
185        assert_eq!(context.channel_id(), Some("embedded-session-1"));
186        assert!(context.is_embedded());
187        assert!(context.validate().is_ok());
188    }
189
190    #[test]
191    fn missing_network_facts_fail_closed() {
192        assert!(RemotingAuthContext::default().validate().is_err());
193        assert!(RemotingAuthContext::new(Some("192.0.2.10".to_owned()), None)
194            .validate()
195            .is_err());
196    }
197
198    #[test]
199    fn mismatched_network_peer_and_session_source_fail_closed() {
200        let peer: SocketAddr = "192.0.2.10:10911".parse().unwrap();
201        let session_source: SocketAddr = "192.0.2.11:10911".parse().unwrap();
202
203        assert!(validate_network_origin(peer, session_source).is_err());
204        assert!(validate_network_origin(peer, peer).is_ok());
205    }
206
207    #[test]
208    fn broker_proxy_is_an_explicitly_trusted_embedded_caller() {
209        assert!(validate_embedded_caller(EmbeddedCaller::BrokerProxy).is_ok());
210    }
211}
rocketmq_auth/remoting_auth_context.rs

rocketmq_auth/
remoting_auth_context.rs
