CASBIN_USER_NAME_HEADERS configuration for audit logging with user name

README.md

@@ -32,6 +32,9 @@ app = Flask(__name__)
 app.config['CASBIN_MODEL'] = 'casbinmodel.conf'
 # Set headers where owner for enforcement policy should be located
 app.config['CASBIN_OWNER_HEADERS'] = {'X-User', 'X-Group'}
+# Add User Audit Logging with user name associated to log
+# i.e. `[2020-11-10 12:55:06,060] ERROR in casbin_enforcer: Unauthorized attempt: method: GET resource: /api/v1/item by user: janedoe@example.com`
+app.config['CASBIN_USER_NAME_HEADERS'] = {'X-User'}
 # Set up Casbin Adapter
 adapter = FileAdapter('rbac_policy.csv')
 casbin_enforcer = CasbinEnforcer(app, adapter)

flask_authz/casbin_enforcer.py

@@ -28,6 +28,7 @@ def __init__(self, app, adapter, watcher=None):
         if watcher:
             self.e.set_watcher(watcher)
         self._owner_loader = None
+        self.user_name_headers = app.config.get("CASBIN_USER_NAME_HEADERS", None)
 
     def set_watcher(self, watcher):
         """
@@ -55,6 +56,9 @@ def enforcer(self, func):
         def wrapper(*args, **kwargs):
             if self.e.watcher and self.e.watcher.should_reload():
                 self.e.watcher.update_callback()
+            # String used to hold the owners user name for audit logging
+            owner_audit = ""
+
             # Check sub, obj act against Casbin polices
             self.app.logger.debug(
                 "Enforce Headers Config: %s\nRequest Headers: %s"
@@ -70,10 +74,10 @@ def wrapper(*args, **kwargs):
                             owner.strip('"'), uri, request.method
                     ):
                         return func(*args, **kwargs)
-            for header in self.app.config.get("CASBIN_OWNER_HEADERS"):
+            for header in map(str.lower, self.app.config.get("CASBIN_OWNER_HEADERS")):
                 if header in request.headers:
                     # Make Authorization Header Parser standard
-                    if header == "Authorization":
+                    if header == "authorization":
                         # Get Auth Value then decode and parse for owner
                         try:
                             owner = authorization_decoder(request.headers.get(header))
@@ -85,6 +89,9 @@ def wrapper(*args, **kwargs):
                                 "decoding is unsupported by flask-casbin at this time"
                             )
                             continue
+
+                        if self.user_name_headers and header in self.user_name_headers:
+                            owner_audit = owner
                         if self.e.enforce(owner, uri, request.method):
                             return func(*args, **kwargs)
                     else:
@@ -97,11 +104,20 @@ def wrapper(*args, **kwargs):
                                 "Enforce against owner: %s header: %s"
                                 % (owner.strip('"'), header)
                             )
+                            if self.user_name_headers and header in map(str.lower, self.user_name_headers):
+                                owner_audit = owner
                             if self.e.enforce(
                                 owner.strip('"'), uri, request.method
                             ):
                                 return func(*args, **kwargs)
             else:
+                self.app.logger.error(
+                    "Unauthorized attempt: method: %s resource: %s%s" % (
+                        request.method,
+                        uri,
+                        "" if not self.user_name_headers and owner_audit != "" else " by user: %s" % owner_audit
+                    )
+                )
                 return (jsonify({"message": "Unauthorized"}), 401)
 
         return wrapper

tests/test_casbin_enforcer.py

@@ -44,24 +44,32 @@ def update_callback(self):
 
 
 @pytest.mark.parametrize(
-    "header, user, method, status",
+    "header, user, method, status, user_name",
     [
-        ("X-User", "alice", "GET", 200),
-        ("X-User", "alice", "POST", 201),
-        ("X-User", "alice", "DELETE", 202),
-        ("X-User", "bob", "GET", 200),
-        ("X-User", "bob", "POST", 401),
-        ("X-User", "bob", "DELETE", 401),
-        ("X-Idp-Groups", "admin", "GET", 401),
-        ("X-Idp-Groups", "users", "GET", 200),
-        ("X-Idp-Groups", "noexist,testnoexist,users", "GET", 200),
-        ("X-Idp-Groups", "noexist testnoexist users", "GET", 200),
-        ("X-Idp-Groups", "noexist, testnoexist, users", "GET", 200),
-        ("Authorization", "Basic Ym9iOnBhc3N3b3Jk", "GET", 200),
-        ("Authorization", "Unsupported Ym9iOnBhc3N3b3Jk", "GET", 401),
+        ("X-User", "alice", "GET", 200, "X-User"),
+        ("X-USER", "alice", "GET", 200, "x-user"),
+        ("x-user", "alice", "GET", 200, "X-USER"),
+        ("X-User", "alice", "GET", 200, "X-USER"),
+        ("X-User", "alice", "GET", 200, "X-Not-A-Header"),
+        ("X-User", "alice", "POST", 201, None),
+        ("X-User", "alice", "DELETE", 202, None),
+        ("X-User", "bob", "GET", 200, None),
+        ("X-User", "bob", "POST", 401, None),
+        ("X-User", "bob", "DELETE", 401, None),
+        ("X-Idp-Groups", "admin", "GET", 401, "X-User"),
+        ("X-Idp-Groups", "users", "GET", 200, None),
+        ("X-Idp-Groups", "noexist,testnoexist,users", "GET", 200, None),
+        ("X-Idp-Groups", "noexist testnoexist users", "GET", 200, None),
+        ("X-Idp-Groups", "noexist, testnoexist, users", "GET", 200, None),
+        ("Authorization", "Basic Ym9iOnBhc3N3b3Jk", "GET", 200, "Authorization"),
+        ("Authorization", "Unsupported Ym9iOnBhc3N3b3Jk", "GET", 401, None),
     ],
 )
-def test_enforcer(app_fixture, enforcer, header, user, method, status):
+def test_enforcer(app_fixture, enforcer, header, user, method, status, user_name):
+    # enable auditing with user name
+    if user_name:
+        enforcer.user_name_headers = {user_name}
+
     @app_fixture.route("/")
     @enforcer.enforcer
     def index():