Merge pull request #15 from zazayaya/master

add owner loader

Author: jessecooper

flask_authz/casbin_enforcer.py

@@ -27,6 +27,7 @@ def __init__(self, app, adapter, watcher=None):
         self.e = casbin.Enforcer(app.config.get("CASBIN_MODEL"), self.adapter, True)
         if watcher:
             self.e.set_watcher(watcher)
+        self._owner_loader = None
 
     def set_watcher(self, watcher):
         """
@@ -38,6 +39,17 @@ def set_watcher(self, watcher):
         """
         self.e.set_watcher(watcher)
 
+    def owner_loader(self, callback):
+        """
+        This sets the callback for get owner. The
+        function return a owner object, or ``None``
+
+        :param callback: The callback for retrieving a owner object.
+        :type callback: callable
+        """
+        self._owner_loader = callback
+        return callback
+
     def enforcer(self, func):
         @wraps(func)
         def wrapper(*args, **kwargs):
@@ -50,6 +62,14 @@ def wrapper(*args, **kwargs):
             )
             # Set resource URI from request
             uri = str(request.path)
+            # Get owner from owner_loader
+            if self._owner_loader:
+                self.app.logger.info("Get owner from owner_loader")
+                for owner in self._owner_loader():
+                    if self.e.enforce(
+                            owner.strip('"'), uri, request.method
+                    ):
+                        return func(*args, **kwargs)
             for header in self.app.config.get("CASBIN_OWNER_HEADERS"):
                 if header in request.headers:
                     # Make Authorization Header Parser standard

tests/test_casbin_enforcer.py

@@ -147,3 +147,47 @@ def test_enforcer_set_watcher(enforcer, watcher):
     assert enforcer.e.watcher is None
     enforcer.set_watcher(watcher())
     assert isinstance(enforcer.e.watcher, watcher)
+
+
+@pytest.mark.parametrize(
+    "owner, method, status",
+    [
+        (["alice"], "GET", 200),
+        (["alice"], "POST", 201),
+        (["alice"], "DELETE", 202),
+        (["bob"], "GET", 200),
+        (["bob"], "POST", 401),
+        (["bob"], "DELETE", 401),
+        (["admin"], "GET", 401),
+        (["users"], "GET", 200),
+        (["alice", "bob"], "POST", 201),
+        (["noexist", "testnoexist"], "POST", 401),
+    ],
+)
+def test_enforcer_with_owner_loader(app_fixture, enforcer, owner, method, status):
+    @app_fixture.route("/")
+    @enforcer.enforcer
+    def index():
+        return jsonify({"message": "passed"}), 200
+
+    @app_fixture.route("/item", methods=["GET", "POST", "DELETE"])
+    @enforcer.enforcer
+    def item():
+        if request.method == "GET":
+            return jsonify({"message": "passed"}), 200
+        elif request.method == "POST":
+            return jsonify({"message": "passed"}), 201
+        elif request.method == "DELETE":
+            return jsonify({"message": "passed"}), 202
+
+    @enforcer.owner_loader
+    def owner_loader():
+        return owner
+
+    c = app_fixture.test_client()
+    # c.post('/add', data=dict(title='2nd Item', text='The text'))
+    rv = c.get("/")
+    assert rv.status_code == 401
+    caller = getattr(c, method.lower())
+    rv = caller("/item")
+    assert rv.status_code == status