rmqtt_auth_http/
lib.rs

1//! HTTP-based authentication and ACL plugin for RMQTT.
2//!
3//! Delegates client authentication and publish/subscribe ACL checks
4//! to an external HTTP API. Supports flexible auth logic based on
5//! custom HTTP endpoint responses.
6//!
7//! # Architecture
8//!
9//! - Hooks into `ClientAuthenticate`, `ClientSubscribeCheckAcl`,
10//!   and `MessagePublishCheckAcl` events.
11//! - Sends HTTP POST requests to configurable URLs with client
12//!   credentials and connection metadata as JSON payload.
13//! - Caches authentication results with configurable TTL to reduce
14//!   HTTP call overhead.
15//! - Supports username/password, client ID, and certificate-based
16//!   authentication flows.
17//!
18#![deny(unsafe_code)]
19
20use std::sync::Arc;
21use std::time::Duration;
22
23use anyhow::anyhow;
24use async_trait::async_trait;
25use bytestring::ByteString;
26use reqwest::{
27    header::{HeaderMap, CONTENT_TYPE},
28    Method, Response, Url,
29};
30use serde::ser::Serialize;
31use tokio::sync::RwLock;
32
33use rmqtt::{
34    acl::{AuthInfo, Rule},
35    codec::v5::SubscribeAckReason,
36    context::ServerContext,
37    hook::{Handler, HookResult, Parameter, Register, ReturnType, Type},
38    macros::Plugin,
39    plugin::{PackageInfo, Plugin},
40    register,
41    types::{
42        AuthResult, ConnectInfo, DashMap, Disconnect, Id, Message, Password, PublishAclResult, Reason,
43        SubscribeAclResult, Superuser, TimestampMillis, TopicName,
44    },
45    utils::timestamp_millis,
46    Error, Result,
47};
48
49use config::PluginConfig;
50
51mod config;
52
53type HashMap<K, V> = std::collections::HashMap<K, V, ahash::RandomState>;
54
55const CACHEABLE: &str = "X-Cache";
56const SUPERUSER: &str = "X-Superuser";
57// const CACHE_KEY: &str = "ACL-CACHE-MAP";
58
59#[derive(Clone, Debug)]
60struct ResponseResult {
61    permission: Permission,
62    superuser: Superuser,
63    cacheable: Cacheable,
64    expire_at: Option<Duration>,
65    acl_data: Option<serde_json::Value>,
66}
67
68impl ResponseResult {
69    #[inline]
70    fn new(permission: Permission, superuser: Superuser, cacheable: Cacheable) -> ResponseResult {
71        ResponseResult { permission, superuser, cacheable, expire_at: None, acl_data: None }
72    }
73}
74
75#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
76enum Permission {
77    Allow(Superuser),
78    Deny,
79    Ignore,
80}
81
82impl TryFrom<(&str, Superuser)> for Permission {
83    type Error = Error;
84
85    #[inline]
86    fn try_from((s, superuser): (&str, Superuser)) -> std::result::Result<Self, Self::Error> {
87        match s {
88            "allow" => Ok(Permission::Allow(superuser)),
89            "deny" => Ok(Permission::Deny),
90            "ignore" => Ok(Permission::Ignore),
91            _ => Err(anyhow!(
92                "The authentication result is incorrect; only 'allow,' 'deny,' or 'ignore' are permitted.",
93            )),
94        }
95    }
96}
97
98impl Permission {
99    #[inline]
100    fn from(s: &str, superuser: Superuser) -> Self {
101        match s {
102            "allow" => Permission::Allow(superuser),
103            "deny" => Permission::Deny,
104            "ignore" => Permission::Ignore,
105            _ => Permission::Allow(superuser),
106        }
107    }
108}
109
110type Cacheable = Option<i64>;
111
112#[derive(Clone, Debug, PartialEq, Eq, Hash, Copy)]
113enum ACLType {
114    Sub = 1,
115    Pub = 2,
116}
117
118impl ACLType {
119    fn as_str(&self) -> &str {
120        match self {
121            Self::Sub => "1",
122            Self::Pub => "2",
123        }
124    }
125}
126
127register!(AuthHttpPlugin::new);
128
129type Caches = Arc<DashMap<Id, std::collections::BTreeMap<TopicName, (Permission, TimestampMillis)>>>;
130
131#[derive(Plugin)]
132struct AuthHttpPlugin {
133    scx: ServerContext,
134    httpc: reqwest::Client,
135    register: Box<dyn Register>,
136    cfg: Arc<RwLock<PluginConfig>>,
137    caches: Caches,
138}
139
140impl AuthHttpPlugin {
141    #[inline]
142    async fn new<S: Into<String>>(scx: ServerContext, name: S) -> Result<Self> {
143        let name = name.into();
144        let cfg = Arc::new(RwLock::new(scx.plugins.read_config::<PluginConfig>(&name)?));
145        log::debug!("{} AuthHttpPlugin cfg: {:?}", name, cfg.read().await);
146        let register = scx.extends.hook_mgr().register();
147        let caches = Arc::new(DashMap::default());
148        let httpc = new_reqwest_client()?;
149        Ok(Self { scx, httpc, register, cfg, caches })
150    }
151}
152
153#[async_trait]
154impl Plugin for AuthHttpPlugin {
155    #[inline]
156    async fn init(&mut self) -> Result<()> {
157        log::info!("{} init", self.name());
158        let cfg = &self.cfg;
159
160        let priority = cfg.read().await.priority;
161        self.register
162            .add_priority(
163                Type::ClientAuthenticate,
164                priority,
165                Box::new(AuthHandler::new(&self.scx, self.httpc.clone(), cfg, &self.caches)),
166            )
167            .await;
168        self.register
169            .add_priority(
170                Type::ClientSubscribeCheckAcl,
171                priority,
172                Box::new(AuthHandler::new(&self.scx, self.httpc.clone(), cfg, &self.caches)),
173            )
174            .await;
175        self.register
176            .add_priority(
177                Type::MessagePublishCheckAcl,
178                priority,
179                Box::new(AuthHandler::new(&self.scx, self.httpc.clone(), cfg, &self.caches)),
180            )
181            .await;
182        self.register
183            .add(
184                Type::ClientKeepalive,
185                Box::new(AuthHandler::new(&self.scx, self.httpc.clone(), cfg, &self.caches)),
186            )
187            .await;
188
189        self.register
190            .add(
191                Type::ClientDisconnected,
192                Box::new(AuthHandler::new(&self.scx, self.httpc.clone(), cfg, &self.caches)),
193            )
194            .await;
195        Ok(())
196    }
197
198    #[inline]
199    async fn get_config(&self) -> Result<serde_json::Value> {
200        self.cfg.read().await.to_json()
201    }
202
203    #[inline]
204    async fn load_config(&mut self) -> Result<()> {
205        let new_cfg = self.scx.plugins.read_config::<PluginConfig>(self.name())?;
206        *self.cfg.write().await = new_cfg;
207        log::debug!("load_config ok,  {:?}", self.cfg);
208        Ok(())
209    }
210
211    #[inline]
212    async fn start(&mut self) -> Result<()> {
213        log::info!("{} start", self.name());
214        self.register.start().await;
215        Ok(())
216    }
217
218    #[inline]
219    async fn stop(&mut self) -> Result<bool> {
220        log::info!("{} stop", self.name());
221        self.register.stop().await;
222        Ok(true)
223    }
224
225    #[inline]
226    async fn attrs(&self) -> serde_json::Value {
227        let mut stats = HashMap::default();
228        for (i, c) in self.caches.iter().enumerate() {
229            if i < 1000 {
230                stats.insert(c.key().to_string(), c.value().len());
231            }
232        }
233
234        serde_json::json!({
235            "caches": self.caches.len(),
236            "stats": stats,
237        })
238    }
239}
240
241struct AuthHandler {
242    scx: ServerContext,
243    httpc: reqwest::Client,
244    cfg: Arc<RwLock<PluginConfig>>,
245    caches: Caches,
246}
247
248impl AuthHandler {
249    fn new(
250        scx: &ServerContext,
251        httpc: reqwest::Client,
252        cfg: &Arc<RwLock<PluginConfig>>,
253        caches: &Caches,
254    ) -> Self {
255        Self { scx: scx.clone(), httpc, cfg: cfg.clone(), caches: caches.clone() }
256    }
257
258    async fn response_result(resp: Response) -> Result<ResponseResult> {
259        if resp.status().is_success() {
260            let content_type = resp.headers().get(CONTENT_TYPE);
261            let is_json_content_type =
262                content_type.map(|hv| hv.as_bytes().starts_with(b"application/json")).unwrap_or_default();
263            log::debug!("content_type: {content_type:?}");
264            log::debug!("is_json_content_type: {is_json_content_type}");
265            let superuser = resp.headers().contains_key(SUPERUSER);
266            // let acl = resp.headers().contains_key(ACL);
267            let cache_timeout = if let Some(tm) = resp.headers().get(CACHEABLE).and_then(|v| v.to_str().ok())
268            {
269                match tm.parse::<i64>() {
270                    Ok(tm) => Some(tm),
271                    Err(e) => {
272                        log::warn!("Parse X-Cache error, {e}");
273                        None
274                    }
275                }
276            } else {
277                None
278            };
279            log::debug!("Cache timeout is {cache_timeout:?}");
280            let resp = if is_json_content_type {
281                let mut body: serde_json::Value = resp.json().await?;
282                log::debug!("body: {body:?}");
283                if let Some(obj) = body.as_object_mut() {
284                    let result = obj
285                        .get("result")
286                        .and_then(|res| res.as_str())
287                        .ok_or_else(|| anyhow!("Authentication result does not exist"))?;
288                    let superuser = obj.get("superuser").and_then(|res| res.as_bool()).unwrap_or(superuser);
289                    let expire_at =
290                        obj.get("expire_at").and_then(|res| res.as_u64().map(Duration::from_secs));
291                    let permission = Permission::try_from((result, superuser))?;
292                    let acl_data = obj.remove("acl");
293
294                    ResponseResult { permission, superuser, cacheable: cache_timeout, expire_at, acl_data }
295                } else if let Some(body) = body.as_str() {
296                    log::debug!("body: {body:?}");
297                    ResponseResult::new(Permission::try_from((body, superuser))?, superuser, cache_timeout)
298                } else {
299                    return Err(anyhow!(format!("The response result is incorrect, {}", body)));
300                }
301            } else {
302                let body = resp.text().await?;
303                log::debug!("body: {body:?}");
304                ResponseResult::new(Permission::from(body.as_str(), superuser), superuser, cache_timeout)
305            };
306            Ok(resp)
307        } else {
308            Ok(ResponseResult::new(Permission::Ignore, false, None))
309        }
310    }
311
312    async fn http_get_request<T: Serialize + ?Sized>(
313        httpc: &reqwest::Client,
314        url: Url,
315        body: &T,
316        headers: HeaderMap,
317        timeout: Duration,
318    ) -> Result<ResponseResult> {
319        log::debug!("http_get_request, timeout: {timeout:?}, url: {url}");
320        match httpc.get(url).headers(headers).timeout(timeout).query(body).send().await {
321            Err(e) => {
322                log::warn!("{e}");
323                Err(anyhow!(e))
324            }
325            Ok(resp) => Self::response_result(resp).await,
326        }
327    }
328
329    async fn http_form_request<T: Serialize + ?Sized>(
330        httpc: &reqwest::Client,
331        url: Url,
332        method: Method,
333        body: &T,
334        headers: HeaderMap,
335        timeout: Duration,
336    ) -> Result<ResponseResult> {
337        log::debug!("http_form_request, method: {method:?}, timeout: {timeout:?}, url: {url}");
338        match httpc.request(method, url).headers(headers).timeout(timeout).form(body).send().await {
339            Err(e) => {
340                log::warn!("{e}");
341                Err(anyhow!(e))
342            }
343            Ok(resp) => Self::response_result(resp).await,
344        }
345    }
346
347    async fn http_json_request<T: Serialize + ?Sized>(
348        httpc: &reqwest::Client,
349        url: Url,
350        method: Method,
351        body: &T,
352        headers: HeaderMap,
353        timeout: Duration,
354    ) -> Result<ResponseResult> {
355        log::debug!("http_json_request, method: {method:?}, timeout: {timeout:?}, url: {url}");
356        match httpc.request(method, url).headers(headers).timeout(timeout).json(body).send().await {
357            Err(e) => {
358                log::warn!("{e}");
359                Err(anyhow!(e))
360            }
361            Ok(resp) => Self::response_result(resp).await,
362        }
363    }
364
365    fn replaces(
366        params: &mut HashMap<String, String>,
367        id: &Id,
368        password: Option<&Password>,
369        protocol: Option<u8>,
370        sub_or_pub: Option<(ACLType, &TopicName)>,
371    ) -> Result<()> {
372        let password =
373            if let Some(p) = password { ByteString::try_from(p.clone())? } else { ByteString::default() };
374        let client_id = id.client_id.as_ref();
375        let username = id.username.as_ref().map(|n| n.as_ref()).unwrap_or("");
376        let remote_addr = id.remote_addr.map(|addr| addr.ip().to_string()).unwrap_or_default();
377        for v in params.values_mut() {
378            *v = v.replace("%u", username);
379            *v = v.replace("%c", client_id);
380            *v = v.replace("%a", &remote_addr);
381            *v = v.replace("%P", &password);
382            if let Some(protocol) = protocol {
383                let mut buffer = itoa::Buffer::new();
384                *v = v.replace("%r", buffer.format(protocol));
385            }
386            if let Some((ref acl_type, topic)) = sub_or_pub {
387                *v = v.replace("%A", acl_type.as_str());
388                *v = v.replace("%t", topic);
389            } else {
390                *v = v.replace("%A", "");
391                *v = v.replace("%t", "");
392            }
393        }
394        Ok(())
395    }
396
397    async fn request(
398        &self,
399        id: &Id,
400        mut req_cfg: config::Req,
401        password: Option<&Password>,
402        protocol: Option<u8>,
403        sub_or_pub: Option<(ACLType, &TopicName)>,
404    ) -> Result<ResponseResult> {
405        log::debug!("{:?} req_cfg.url.path(): {:?}", id, req_cfg.url.path());
406        let (headers, timeout) = {
407            let cfg = self.cfg.read().await;
408            let headers = match (cfg.headers(), req_cfg.headers()) {
409                (Some(def_headers), Some(req_headers)) => {
410                    let mut headers = def_headers.clone();
411                    headers.extend(req_headers.clone());
412                    headers
413                }
414                (Some(def_headers), None) => def_headers.clone(),
415                (None, Some(req_headers)) => req_headers.clone(),
416                (None, None) => HeaderMap::new(),
417            };
418            (headers, cfg.http_timeout)
419        };
420
421        let auth_result = if req_cfg.is_get() {
422            let body = &mut req_cfg.params;
423            Self::replaces(body, id, password, protocol, sub_or_pub)?;
424            Self::http_get_request(&self.httpc, req_cfg.url, body, headers, timeout).await?
425        } else if req_cfg.json_body() {
426            let body = &mut req_cfg.params;
427            Self::replaces(body, id, password, protocol, sub_or_pub)?;
428            Self::http_json_request(&self.httpc, req_cfg.url, req_cfg.method, body, headers, timeout).await?
429        } else {
430            //form body
431            let body = &mut req_cfg.params;
432            Self::replaces(body, id, password, protocol, sub_or_pub)?;
433            Self::http_form_request(&self.httpc, req_cfg.url, req_cfg.method, body, headers, timeout).await?
434        };
435        log::debug!("auth_result: {auth_result:?}");
436        Ok(auth_result)
437    }
438
439    #[inline]
440    async fn auth(&self, connect_info: &ConnectInfo) -> (Permission, Option<AuthInfo>) {
441        if let Some(req) = { self.cfg.read().await.http_auth_req.clone() } {
442            match self
443                .request(
444                    connect_info.id(),
445                    req,
446                    connect_info.password(),
447                    Some(connect_info.proto_ver()),
448                    None,
449                )
450                .await
451            {
452                Ok(auth_res) => {
453                    log::debug!("auth result: {auth_res:?}");
454                    let auth_info = if matches!(auth_res.permission, Permission::Allow(_)) {
455                        if let Some(acl_data) =
456                            auth_res.acl_data.as_ref().and_then(|acl_data| acl_data.as_array())
457                        {
458                            match acl_data
459                                .iter()
460                                .map(|acl| Rule::try_from((acl, connect_info)))
461                                .collect::<Result<Vec<Rule>>>()
462                            {
463                                Ok(rules) => {
464                                    let auth_info = AuthInfo {
465                                        superuser: auth_res.superuser,
466                                        expire_at: auth_res.expire_at,
467                                        rules,
468                                    };
469                                    log::debug!("auth_info: {auth_info:?}");
470                                    Some(auth_info)
471                                }
472                                Err(e) => {
473                                    log::warn!("{} {}", connect_info.id(), e);
474                                    None
475                                }
476                            }
477                        } else {
478                            None
479                        }
480                    } else {
481                        None
482                    };
483                    (auth_res.permission, auth_info)
484                }
485                Err(e) => {
486                    log::warn!("{:?} auth error, {:?}", connect_info.id(), e);
487                    if self.cfg.read().await.deny_if_error {
488                        (Permission::Deny, None)
489                    } else {
490                        (Permission::Ignore, None)
491                    }
492                }
493            }
494        } else {
495            (Permission::Ignore, None)
496        }
497    }
498
499    #[inline]
500    async fn acl(
501        &self,
502        id: &Id,
503        protocol: Option<u8>,
504        sub_or_pub: Option<(ACLType, &TopicName)>,
505    ) -> (Permission, Cacheable) {
506        if let Some(req) = { self.cfg.read().await.http_acl_req.clone() } {
507            match self.request(id, req, None, protocol, sub_or_pub).await {
508                Ok(acl_res) => {
509                    log::debug!("acl result: {acl_res:?}");
510                    (acl_res.permission, acl_res.cacheable)
511                }
512                Err(e) => {
513                    log::warn!("{id:?} acl error, {e}");
514                    if self.cfg.read().await.deny_if_error {
515                        (Permission::Deny, None)
516                    } else {
517                        (Permission::Ignore, None)
518                    }
519                }
520            }
521        } else {
522            (Permission::Ignore, None)
523        }
524    }
525
526    #[inline]
527    fn cache_set(&self, id: Id, topic: TopicName, perm: Permission, expire: TimestampMillis) {
528        self.caches.entry(id).or_default().insert(topic, (perm, expire));
529    }
530
531    #[inline]
532    fn cache_get(&self, id: &Id, topic: &TopicName) -> Option<(Permission, TimestampMillis)> {
533        self.caches.get(id).and_then(|c| c.get(topic).map(|(perm, expire)| (*perm, *expire)))
534    }
535
536    #[inline]
537    fn cache_remove(&self, id: &Id) {
538        self.caches.remove(id);
539    }
540}
541
542#[async_trait]
543impl Handler for AuthHandler {
544    async fn hook(&self, param: &Parameter, acc: Option<HookResult>) -> ReturnType {
545        match param {
546            Parameter::ClientAuthenticate(connect_info) => {
547                log::debug!("ClientAuthenticate auth-http");
548                if matches!(
549                    acc,
550                    Some(HookResult::AuthResult(AuthResult::BadUsernameOrPassword))
551                        | Some(HookResult::AuthResult(AuthResult::NotAuthorized))
552                ) {
553                    return (false, acc);
554                }
555
556                return match self.auth(connect_info).await {
557                    (Permission::Allow(superuser), auth_info) => {
558                        if auth_info.as_ref().map(|ai| ai.is_expired()).unwrap_or_default() {
559                            log::warn!("{} authentication information has expired.", connect_info.id());
560                            (false, Some(HookResult::AuthResult(AuthResult::NotAuthorized)))
561                        } else {
562                            (false, Some(HookResult::AuthResult(AuthResult::Allow(superuser, auth_info))))
563                        }
564                    }
565                    (Permission::Deny, _) => {
566                        (false, Some(HookResult::AuthResult(AuthResult::BadUsernameOrPassword)))
567                    }
568                    (Permission::Ignore, _) => (true, None),
569                };
570            }
571
572            Parameter::ClientSubscribeCheckAcl(session, subscribe) => {
573                if let Some(HookResult::SubscribeAclResult(acl_result)) = &acc {
574                    if acl_result.failure() {
575                        return (false, acc);
576                    }
577                }
578
579                if let Some(auth_info) = &session.auth_info {
580                    if let Some(acl_res) = auth_info.subscribe_acl(subscribe).await {
581                        return acl_res;
582                    }
583                }
584
585                //Permission, Cacheable
586                let (acl_res, _) = self
587                    .acl(
588                        &session.id,
589                        session.protocol().await.ok(),
590                        Some((ACLType::Sub, &subscribe.topic_filter)),
591                    )
592                    .await;
593                return match acl_res {
594                    Permission::Allow(_) => (
595                        false,
596                        Some(HookResult::SubscribeAclResult(SubscribeAclResult::new_success(
597                            subscribe.opts.qos(),
598                            None,
599                        ))),
600                    ),
601                    Permission::Deny => (
602                        false,
603                        Some(HookResult::SubscribeAclResult(SubscribeAclResult::new_failure(
604                            SubscribeAckReason::NotAuthorized,
605                        ))),
606                    ),
607                    Permission::Ignore => (true, None),
608                };
609            }
610
611            Parameter::MessagePublishCheckAcl(session, publish) => {
612                log::debug!("MessagePublishCheckAcl");
613                if let Some(HookResult::PublishAclResult(acl_res)) = &acc {
614                    if acl_res.is_rejected() {
615                        return (false, acc);
616                    }
617                }
618
619                if let Some(auth_info) = &session.auth_info {
620                    if let Some(acl_res) =
621                        auth_info.publish_acl(publish, self.cfg.read().await.disconnect_if_pub_rejected).await
622                    {
623                        return acl_res;
624                    }
625                }
626
627                let acl_res = if let Some((acl_res, expire)) = self.cache_get(&session.id, &publish.topic) {
628                    if expire < 0 || timestamp_millis() < expire {
629                        Some(acl_res)
630                    } else {
631                        None
632                    }
633                } else {
634                    None
635                };
636
637                let acl_res = if let Some(acl_res) = acl_res {
638                    acl_res
639                } else {
640                    //Permission, Cacheable
641                    let (acl_res, cacheable) = self
642                        .acl(&session.id, session.protocol().await.ok(), Some((ACLType::Pub, &publish.topic)))
643                        .await;
644                    if let Some(tm) = cacheable {
645                        let expire = if tm < 0 { tm } else { timestamp_millis() + tm };
646
647                        self.cache_set(session.id.clone(), publish.topic.clone(), acl_res, expire);
648                    }
649                    acl_res
650                };
651
652                return match acl_res {
653                    Permission::Allow(_) => {
654                        (false, Some(HookResult::PublishAclResult(PublishAclResult::allow())))
655                    }
656                    Permission::Deny => (
657                        false,
658                        Some(HookResult::PublishAclResult(PublishAclResult::rejected(
659                            self.cfg.read().await.disconnect_if_pub_rejected,
660                            None,
661                        ))),
662                    ),
663                    Permission::Ignore => (true, None),
664                };
665            }
666
667            Parameter::ClientKeepalive(s, _) => {
668                if let Some(auth) = &s.auth_info {
669                    log::debug!("Keepalive auth-http, is_expired: {:?}", auth.is_expired());
670                    if auth.is_expired() && self.cfg.read().await.disconnect_if_expiry {
671                        if let Some(tx) = self.scx.extends.shared().await.entry(s.id().clone()).tx() {
672                            if let Err(e) = tx.unbounded_send(Message::Closed(Reason::ConnectDisconnect(
673                                Some(Disconnect::Other("Http Auth expired".into())),
674                            ))) {
675                                log::warn!("{} {}", s.id(), e);
676                            }
677                        }
678                    }
679                }
680            }
681
682            Parameter::ClientDisconnected(s, _) => {
683                self.cache_remove(&s.id);
684            }
685
686            _ => {
687                log::error!("unimplemented, {param:?}")
688            }
689        }
690        (true, acc)
691    }
692}
693
694fn new_reqwest_client() -> Result<reqwest::Client> {
695    reqwest::Client::builder()
696        .connect_timeout(Duration::from_secs(10))
697        .timeout(Duration::from_secs(10))
698        .build()
699        .map_err(|e| anyhow!(e))
700}
rmqtt_auth_http/lib.rs

rmqtt_auth_http/
lib.rs
