sqlx_adapter/
adapter.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use async_trait::async_trait;
19use casbin::{error::AdapterError, Adapter, Error as CasbinError, Filter, Model, Result};
20use std::sync::{
21    atomic::{AtomicBool, Ordering},
22    Arc,
23};
24
25use crate::{error::*, models::*};
26
27use crate::actions as adapter;
28
29#[cfg(feature = "mysql")]
30use sqlx::mysql::MySqlPoolOptions;
31#[cfg(feature = "postgres")]
32use sqlx::postgres::PgPoolOptions;
33#[cfg(feature = "sqlite")]
34use sqlx::sqlite::SqlitePoolOptions;
35
36#[derive(Clone)]
37pub struct SqlxAdapter {
38    pool: adapter::ConnectionPool,
39    is_filtered: Arc<AtomicBool>,
40}
41
42//pub const TABLE_NAME: &str = "casbin_rule";
43
44impl<'a> SqlxAdapter {
45    pub async fn new<U: Into<String>>(url: U, pool_size: u32) -> Result<Self> {
46        #[cfg(feature = "postgres")]
47        let pool = PgPoolOptions::new()
48            .max_connections(pool_size)
49            .connect(&url.into())
50            .await
51            .map_err(|err| CasbinError::from(AdapterError(Box::new(Error::SqlxError(err)))))?;
52
53        #[cfg(feature = "mysql")]
54        let pool = MySqlPoolOptions::new()
55            .max_connections(pool_size)
56            .connect(&url.into())
57            .await
58            .map_err(|err| CasbinError::from(AdapterError(Box::new(Error::SqlxError(err)))))?;
59
60        #[cfg(feature = "sqlite")]
61        let pool = SqlitePoolOptions::new()
62            .max_connections(pool_size)
63            .connect(&url.into())
64            .await
65            .map_err(|err| CasbinError::from(AdapterError(Box::new(Error::SqlxError(err)))))?;
66
67        adapter::new(&pool).await.map(|_| Self {
68            pool,
69            is_filtered: Arc::new(AtomicBool::new(false)),
70        })
71    }
72
73    pub async fn new_with_pool(pool: adapter::ConnectionPool) -> Result<Self> {
74        adapter::new(&pool).await.map(|_| Self {
75            pool,
76            is_filtered: Arc::new(AtomicBool::new(false)),
77        })
78    }
79
80    pub(crate) fn save_policy_line(
81        &self,
82        ptype: &'a str,
83        rule: &'a [String],
84    ) -> Option<NewCasbinRule<'a>> {
85        if ptype.trim().is_empty() || rule.is_empty() {
86            return None;
87        }
88
89        let mut new_rule = NewCasbinRule {
90            ptype,
91            v0: "",
92            v1: "",
93            v2: "",
94            v3: "",
95            v4: "",
96            v5: "",
97        };
98
99        new_rule.v0 = &rule[0];
100
101        if rule.len() > 1 {
102            new_rule.v1 = &rule[1];
103        }
104
105        if rule.len() > 2 {
106            new_rule.v2 = &rule[2];
107        }
108
109        if rule.len() > 3 {
110            new_rule.v3 = &rule[3];
111        }
112
113        if rule.len() > 4 {
114            new_rule.v4 = &rule[4];
115        }
116
117        if rule.len() > 5 {
118            new_rule.v5 = &rule[5];
119        }
120
121        Some(new_rule)
122    }
123
124    pub(crate) fn load_policy_line(&self, casbin_rule: &CasbinRule) -> Option<Vec<String>> {
125        if casbin_rule.ptype.chars().next().is_some() {
126            return self.normalize_policy(casbin_rule);
127        }
128
129        None
130    }
131
132    fn normalize_policy(&self, casbin_rule: &CasbinRule) -> Option<Vec<String>> {
133        let mut result = vec![
134            &casbin_rule.v0,
135            &casbin_rule.v1,
136            &casbin_rule.v2,
137            &casbin_rule.v3,
138            &casbin_rule.v4,
139            &casbin_rule.v5,
140        ];
141
142        while let Some(last) = result.last() {
143            if last.is_empty() {
144                result.pop();
145            } else {
146                break;
147            }
148        }
149
150        if !result.is_empty() {
151            return Some(result.iter().map(|&x| x.to_owned()).collect());
152        }
153
154        None
155    }
156}
157
158#[async_trait]
159impl Adapter for SqlxAdapter {
160    async fn load_policy(&mut self, m: &mut dyn Model) -> Result<()> {
161        let rules = adapter::load_policy(&self.pool).await?;
162
163        for casbin_rule in &rules {
164            let rule = self.load_policy_line(casbin_rule);
165
166            if let Some(ref sec) = casbin_rule.ptype.chars().next().map(|x| x.to_string()) {
167                if let Some(t1) = m.get_mut_model().get_mut(sec) {
168                    if let Some(t2) = t1.get_mut(&casbin_rule.ptype) {
169                        if let Some(rule) = rule {
170                            t2.get_mut_policy().insert(rule);
171                        }
172                    }
173                }
174            }
175        }
176
177        Ok(())
178    }
179
180    async fn load_filtered_policy<'a>(&mut self, m: &mut dyn Model, f: Filter<'a>) -> Result<()> {
181        let rules = adapter::load_filtered_policy(&self.pool, &f).await?;
182        self.is_filtered.store(true, Ordering::SeqCst);
183
184        for casbin_rule in &rules {
185            if let Some(policy) = self.normalize_policy(casbin_rule) {
186                if let Some(ref sec) = casbin_rule.ptype.chars().next().map(|x| x.to_string()) {
187                    if let Some(t1) = m.get_mut_model().get_mut(sec) {
188                        if let Some(t2) = t1.get_mut(&casbin_rule.ptype) {
189                            t2.get_mut_policy().insert(policy);
190                        }
191                    }
192                }
193            }
194        }
195
196        Ok(())
197    }
198
199    async fn save_policy(&mut self, m: &mut dyn Model) -> Result<()> {
200        let mut rules = vec![];
201
202        if let Some(ast_map) = m.get_model().get("p") {
203            for (ptype, ast) in ast_map {
204                let new_rules = ast
205                    .get_policy()
206                    .into_iter()
207                    .filter_map(|x| self.save_policy_line(ptype, x));
208
209                rules.extend(new_rules);
210            }
211        }
212
213        if let Some(ast_map) = m.get_model().get("g") {
214            for (ptype, ast) in ast_map {
215                let new_rules = ast
216                    .get_policy()
217                    .into_iter()
218                    .filter_map(|x| self.save_policy_line(ptype, x));
219
220                rules.extend(new_rules);
221            }
222        }
223        adapter::save_policy(&self.pool, rules).await
224    }
225
226    async fn add_policy(&mut self, _sec: &str, ptype: &str, rule: Vec<String>) -> Result<bool> {
227        if let Some(new_rule) = self.save_policy_line(ptype, rule.as_slice()) {
228            return adapter::add_policy(&self.pool, new_rule).await;
229        }
230
231        Ok(false)
232    }
233
234    async fn add_policies(
235        &mut self,
236        _sec: &str,
237        ptype: &str,
238        rules: Vec<Vec<String>>,
239    ) -> Result<bool> {
240        let new_rules = rules
241            .iter()
242            .filter_map(|x| self.save_policy_line(ptype, x))
243            .collect::<Vec<NewCasbinRule>>();
244
245        adapter::add_policies(&self.pool, new_rules).await
246    }
247
248    async fn remove_policy(&mut self, _sec: &str, pt: &str, rule: Vec<String>) -> Result<bool> {
249        adapter::remove_policy(&self.pool, pt, rule).await
250    }
251
252    async fn remove_policies(
253        &mut self,
254        _sec: &str,
255        pt: &str,
256        rules: Vec<Vec<String>>,
257    ) -> Result<bool> {
258        adapter::remove_policies(&self.pool, pt, rules).await
259    }
260
261    async fn remove_filtered_policy(
262        &mut self,
263        _sec: &str,
264        pt: &str,
265        field_index: usize,
266        field_values: Vec<String>,
267    ) -> Result<bool> {
268        if field_index <= 5 && !field_values.is_empty() && field_values.len() + field_index <= 6 {
269            adapter::remove_filtered_policy(&self.pool, pt, field_index, field_values).await
270        } else {
271            Ok(false)
272        }
273    }
274
275    async fn clear_policy(&mut self) -> Result<()> {
276        adapter::clear_policy(&self.pool).await
277    }
278
279    fn is_filtered(&self) -> bool {
280        self.is_filtered.load(Ordering::SeqCst)
281    }
282}
283
284#[cfg(test)]
285mod tests {
286    use super::*;
287
288    fn to_owned(v: Vec<&str>) -> Vec<String> {
289        v.into_iter().map(|x| x.to_owned()).collect()
290    }
291
292    fn test_database_url() -> String {
293        #[cfg(feature = "postgres")]
294        let default_url = "postgres://casbin_rs:casbin_rs@localhost:5432/casbin";
295
296        #[cfg(feature = "mysql")]
297        let default_url = "mysql://casbin_rs:casbin_rs@localhost:3306/casbin";
298
299        #[cfg(feature = "sqlite")]
300        let default_url = "sqlite:casbin.db";
301
302        std::env::var("DATABASE_URL").unwrap_or_else(|_| default_url.to_owned())
303    }
304
305    #[cfg_attr(feature = "runtime-async-std", async_std::test)]
306    #[cfg_attr(feature = "runtime-tokio", tokio::test(flavor = "multi_thread"))]
307    async fn test_create() {
308        use casbin::prelude::*;
309
310        let m = DefaultModel::from_file("examples/rbac_model.conf")
311            .await
312            .unwrap();
313
314        let adapter = {
315            #[cfg(feature = "postgres")]
316            {
317                SqlxAdapter::new(test_database_url(), 8).await.unwrap()
318            }
319
320            #[cfg(feature = "mysql")]
321            {
322                SqlxAdapter::new(test_database_url(), 8).await.unwrap()
323            }
324
325            #[cfg(feature = "sqlite")]
326            {
327                SqlxAdapter::new(test_database_url(), 8).await.unwrap()
328            }
329        };
330
331        assert!(Enforcer::new(m, adapter).await.is_ok());
332    }
333
334    #[cfg_attr(feature = "runtime-async-std", async_std::test)]
335    #[cfg_attr(feature = "runtime-tokio", tokio::test(flavor = "multi_thread"))]
336    async fn test_create_with_pool() {
337        use casbin::prelude::*;
338
339        let m = DefaultModel::from_file("examples/rbac_model.conf")
340            .await
341            .unwrap();
342        let pool = {
343            #[cfg(feature = "postgres")]
344            {
345                let url = test_database_url();
346                PgPoolOptions::new()
347                    .max_connections(8)
348                    .connect(&url)
349                    .await
350                    .unwrap()
351            }
352
353            #[cfg(feature = "mysql")]
354            {
355                let url = test_database_url();
356                MySqlPoolOptions::new()
357                    .max_connections(8)
358                    .connect(&url)
359                    .await
360                    .unwrap()
361            }
362
363            #[cfg(feature = "sqlite")]
364            {
365                let url = test_database_url();
366                SqlitePoolOptions::new()
367                    .max_connections(8)
368                    .connect(&url)
369                    .await
370                    .unwrap()
371            }
372        };
373
374        let adapter = SqlxAdapter::new_with_pool(pool).await.unwrap();
375
376        assert!(Enforcer::new(m, adapter).await.is_ok());
377    }
378
379    #[cfg_attr(feature = "runtime-async-std", async_std::test)]
380    #[cfg_attr(feature = "runtime-tokio", tokio::test(flavor = "multi_thread"))]
381    async fn test_adapter() {
382        use casbin::prelude::*;
383
384        let file_adapter = FileAdapter::new("examples/rbac_policy.csv");
385
386        let m = DefaultModel::from_file("examples/rbac_model.conf")
387            .await
388            .unwrap();
389
390        let mut e = Enforcer::new(m, file_adapter).await.unwrap();
391        #[cfg(feature = "postgres")]
392        let postgres_pool = {
393            let url = test_database_url();
394            PgPoolOptions::new()
395                .max_connections(8)
396                .connect(&url)
397                .await
398                .unwrap()
399        };
400        let mut adapter = {
401            #[cfg(feature = "postgres")]
402            {
403                SqlxAdapter::new_with_pool(postgres_pool.clone())
404                    .await
405                    .unwrap()
406            }
407
408            #[cfg(feature = "mysql")]
409            {
410                SqlxAdapter::new(test_database_url(), 8).await.unwrap()
411            }
412
413            #[cfg(feature = "sqlite")]
414            {
415                SqlxAdapter::new(test_database_url(), 8).await.unwrap()
416            }
417        };
418
419        assert!(adapter.save_policy(e.get_mut_model()).await.is_ok());
420
421        #[cfg(feature = "postgres")]
422        {
423            let model = DefaultModel::from_file("examples/rbac_model.conf")
424                .await
425                .unwrap();
426            let reloaded_adapter = SqlxAdapter::new_with_pool(postgres_pool).await.unwrap();
427            let reloaded = Enforcer::new(model, reloaded_adapter).await.unwrap();
428
429            assert!(reloaded.enforce(("alice", "data1", "read")).unwrap());
430        }
431
432        assert!(adapter
433            .remove_policy("", "p", to_owned(vec!["alice", "data1", "read"]))
434            .await
435            .unwrap());
436        assert!(adapter
437            .remove_policy("", "p", to_owned(vec!["bob", "data2", "write"]))
438            .await
439            .is_ok());
440        assert!(adapter
441            .remove_policy("", "p", to_owned(vec!["data2_admin", "data2", "read"]))
442            .await
443            .is_ok());
444        assert!(adapter
445            .remove_policy("", "p", to_owned(vec!["data2_admin", "data2", "write"]))
446            .await
447            .is_ok());
448        assert!(adapter
449            .remove_policy("", "g", to_owned(vec!["alice", "data2_admin"]))
450            .await
451            .is_ok());
452
453        assert!(adapter
454            .add_policy("", "p", to_owned(vec!["alice", "data1", "read"]))
455            .await
456            .is_ok());
457        assert!(adapter
458            .add_policy("", "p", to_owned(vec!["bob", "data2", "write"]))
459            .await
460            .is_ok());
461        assert!(adapter
462            .add_policy("", "p", to_owned(vec!["data2_admin", "data2", "read"]))
463            .await
464            .is_ok());
465        assert!(adapter
466            .add_policy("", "p", to_owned(vec!["data2_admin", "data2", "write"]))
467            .await
468            .is_ok());
469
470        assert!(adapter
471            .remove_policies(
472                "",
473                "p",
474                vec![
475                    to_owned(vec!["alice", "data1", "read"]),
476                    to_owned(vec!["bob", "data2", "write"]),
477                    to_owned(vec!["data2_admin", "data2", "read"]),
478                    to_owned(vec!["data2_admin", "data2", "write"]),
479                ]
480            )
481            .await
482            .is_ok());
483
484        assert!(adapter
485            .add_policies(
486                "",
487                "p",
488                vec![
489                    to_owned(vec!["alice", "data1", "read"]),
490                    to_owned(vec!["bob", "data2", "write"]),
491                    to_owned(vec!["data2_admin", "data2", "read"]),
492                    to_owned(vec!["data2_admin", "data2", "write"]),
493                ]
494            )
495            .await
496            .is_ok());
497
498        assert!(adapter
499            .add_policy("", "g", to_owned(vec!["alice", "data2_admin"]))
500            .await
501            .is_ok());
502
503        assert!(adapter
504            .remove_policy("", "p", to_owned(vec!["alice", "data1", "read"]))
505            .await
506            .is_ok());
507        assert!(adapter
508            .remove_policy("", "p", to_owned(vec!["bob", "data2", "write"]))
509            .await
510            .is_ok());
511        assert!(adapter
512            .remove_policy("", "p", to_owned(vec!["data2_admin", "data2", "read"]))
513            .await
514            .is_ok());
515        assert!(adapter
516            .remove_policy("", "p", to_owned(vec!["data2_admin", "data2", "write"]))
517            .await
518            .is_ok());
519        assert!(adapter
520            .remove_policy("", "g", to_owned(vec!["alice", "data2_admin"]))
521            .await
522            .is_ok());
523
524        assert!(!adapter
525            .remove_policy(
526                "",
527                "g",
528                to_owned(vec!["alice", "data2_admin", "not_exists"])
529            )
530            .await
531            .unwrap());
532
533        assert!(adapter
534            .add_policy("", "g", to_owned(vec!["alice", "data2_admin"]))
535            .await
536            .is_ok());
537        assert!(adapter
538            .add_policy("", "g", to_owned(vec!["alice", "data2_admin"]))
539            .await
540            .is_err());
541
542        assert!(!adapter
543            .remove_filtered_policy(
544                "",
545                "g",
546                0,
547                to_owned(vec!["alice", "data2_admin", "not_exists"]),
548            )
549            .await
550            .unwrap());
551
552        assert!(adapter
553            .remove_filtered_policy("", "g", 0, to_owned(vec!["alice", "data2_admin"]))
554            .await
555            .unwrap());
556
557        assert!(adapter
558            .add_policy(
559                "",
560                "g",
561                to_owned(vec!["alice", "data2_admin", "domain1", "domain2"]),
562            )
563            .await
564            .is_ok());
565        assert!(adapter
566            .remove_filtered_policy(
567                "",
568                "g",
569                1,
570                to_owned(vec!["data2_admin", "domain1", "domain2"]),
571            )
572            .await
573            .unwrap());
574
575        // GitHub issue: https://github.com/apache/casbin-sqlx-adapter/issues/64
576        assert!(adapter
577            .add_policy("", "g", to_owned(vec!["carol", "data1_admin"]),)
578            .await
579            .is_ok());
580        assert!(adapter
581            .remove_filtered_policy("", "g", 0, to_owned(vec!["carol"]),)
582            .await
583            .unwrap());
584        assert_eq!(Vec::<String>::new(), e.get_roles_for_user("carol", None));
585
586        // GitHub issue: https://github.com/apache/casbin-sqlx-adapter/pull/90
587        // add policies:
588        // p, alice_rfp, book_rfp, read_rfp
589        // p, bob_rfp, book_rfp, read_rfp
590        // p, bob_rfp, book_rfp, write_rfp
591        // p, alice_rfp, pen_rfp, get_rfp
592        // p, bob_rfp, pen_rfp, get_rfp
593        // p, alice_rfp, pencil_rfp, get_rfp
594        assert!(adapter
595            .add_policy("", "p", to_owned(vec!["alice_rfp", "book_rfp", "read_rfp"]),)
596            .await
597            .is_ok());
598        assert!(adapter
599            .add_policy("", "p", to_owned(vec!["bob_rfp", "book_rfp", "read_rfp"]),)
600            .await
601            .is_ok());
602        assert!(adapter
603            .add_policy("", "p", to_owned(vec!["bob_rfp", "book_rfp", "write_rfp"]),)
604            .await
605            .is_ok());
606        assert!(adapter
607            .add_policy("", "p", to_owned(vec!["alice_rfp", "pen_rfp", "get_rfp"]),)
608            .await
609            .is_ok());
610        assert!(adapter
611            .add_policy("", "p", to_owned(vec!["bob_rfp", "pen_rfp", "get_rfp"]),)
612            .await
613            .is_ok());
614        assert!(adapter
615            .add_policy(
616                "",
617                "p",
618                to_owned(vec!["alice_rfp", "pencil_rfp", "get_rfp"]),
619            )
620            .await
621            .is_ok());
622
623        // should remove (return true) all policies where "book_rfp" is in the second position
624        assert!(adapter
625            .remove_filtered_policy("", "p", 1, to_owned(vec!["book_rfp"]),)
626            .await
627            .unwrap());
628
629        // should remove (return true) all policies which match "alice_rfp" on first position
630        // and "get_rfp" on third position
631        assert!(adapter
632            .remove_filtered_policy("", "p", 0, to_owned(vec!["alice_rfp", "", "get_rfp"]),)
633            .await
634            .unwrap());
635
636        // shadow the previous enforcer
637        let mut e = Enforcer::new(
638            "examples/rbac_with_domains_model.conf",
639            "examples/rbac_with_domains_policy.csv",
640        )
641        .await
642        .unwrap();
643
644        assert!(adapter.save_policy(e.get_mut_model()).await.is_ok());
645        e.set_adapter(adapter).await.unwrap();
646
647        let filter = Filter {
648            p: vec!["", "domain1"],
649            g: vec!["", "", "domain1"],
650        };
651
652        e.load_filtered_policy(filter).await.unwrap();
653        assert!(e.enforce(("alice", "domain1", "data1", "read")).unwrap());
654        assert!(e.enforce(("alice", "domain1", "data1", "write")).unwrap());
655        assert!(!e.enforce(("alice", "domain1", "data2", "read")).unwrap());
656        assert!(!e.enforce(("alice", "domain1", "data2", "write")).unwrap());
657        assert!(!e.enforce(("bob", "domain2", "data2", "read")).unwrap());
658        assert!(!e.enforce(("bob", "domain2", "data2", "write")).unwrap());
659    }
660}
sqlx_adapter/adapter.rs

sqlx_adapter/
adapter.rs
