feat: add initial code

.gitignore

@@ -157,4 +157,4 @@ cython_debug/
 #  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
-#.idea/
+.idea/

README.md

@@ -1 +1,53 @@
-# async-postgres-watcher
+# async-postgres-watcher
+
+[![Discord](https://img.shields.io/discord/1022748306096537660?logo=discord&label=discord&color=5865F2)](https://discord.gg/S5UjpzGZjN)
+
+Async casbin role watcher to be used for monitoring updates to casbin policies
+
+## Basic Usage Example
+
+### With Flask-authz
+
+```python
+from flask_authz import CasbinEnforcer
+from async_postgres_watcher import AsyncPostgresWatcher
+from flask import Flask
+from casbin.persist.adapters import FileAdapter
+
+casbin_enforcer = CasbinEnforcer(app, adapter)
+
+watcher = AsyncPostgresWatcher(host=HOST, port=PORT, user=USER, password=PASSWORD, dbname=DBNAME)
+watcher.set_update_callback(casbin_enforcer.e.load_policy)
+
+casbin_enforcer.set_watcher(watcher)
+```
+
+## Basic Usage Example With SSL Enabled
+
+See [asyncpg documentation](https://magicstack.github.io/asyncpg/current/api/index.html#connection) for full details of SSL parameters.
+
+### With Flask-authz
+
+```python
+from flask_authz import CasbinEnforcer
+from async_postgres_watcher import AsyncPostgresWatcher
+from flask import Flask
+from casbin.persist.adapters import FileAdapter
+
+casbin_enforcer = CasbinEnforcer(app, adapter)
+
+# If check_hostname is True, the SSL context is created with sslmode=verify-full.
+# If check_hostname is False, the SSL context is created with sslmode=verify-ca.
+watcher = AsyncPostgresWatcher(host=HOST, port=PORT, user=USER, password=PASSWORD, dbname=DBNAME, sslrootcert=SSLROOTCERT, check_hostname = True, sslcert=SSLCERT, sslkey=SSLKEY)
+
+watcher.set_update_callback(casbin_enforcer.e.load_policy)
+casbin_enforcer.set_watcher(watcher)
+```
+
+## Getting Help
+
+- [PyCasbin](https://github.com/casbin/pycasbin)
+
+## License
+
+This project is under Apache 2.0 License. See the [LICENSE](LICENSE) file for the full license text.

async_postgres_watcher/__init__.py

@@ -0,0 +1,15 @@
+# Copyright 2024 The casbin Authors. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .watcher import AsyncPostgresWatcher

async_postgres_watcher/watcher.py

@@ -0,0 +1,100 @@
+# Copyright 2024 The casbin Authors. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from typing import Optional, Callable
+import asyncio
+import asyncpg
+import ssl
+import time
+
+POSTGRESQL_CHANNEL_NAME = "casbin_role_watcher"
+
+
+class AsyncPostgresWatcher:
+    def __init__(
+            self,
+            host: str,
+            user: str,
+            password: str,
+            port: Optional[int] = 5432,
+            dbname: Optional[str] = "postgres",
+            channel_name: Optional[str] = POSTGRESQL_CHANNEL_NAME,
+            sslrootcert: Optional[str] = None,
+            # If True, equivalent to sslmode=verify-full, if False: sslmode=verify-ca.
+            check_hostname: Optional[bool] = True,
+            sslcert: Optional[str] = None,
+            sslkey: Optional[str] = None
+    ):
+        self.loop = asyncio.get_event_loop()
+        self.running = True
+        self.callback = None
+        self.host = host
+        self.port = port
+        self.user = user
+        self.password = password
+        self.dbname = dbname
+        self.channel_name = channel_name
+
+        if sslrootcert is not None and sslcert is not None and sslkey is not None:
+            self.sslctx = ssl.create_default_context(
+                ssl.Purpose.SERVER_AUTH,
+                cafile=sslrootcert
+            )
+            self.sslctx.check_hostname = check_hostname
+            self.sslctx.load_cert_chain(sslcert, keyfile=sslkey)
+        else:
+            self.sslctx = False
+
+        self.loop.create_task(self.subscriber())
+
+    async def notify(self, pid, channel, payload):
+        print(f"Notify: {payload}")
+        if self.callback is not None:
+            if asyncio.iscoroutinefunction(self.callback):
+                await self.callback(payload)
+            else:
+                self.callback(payload)
+
+    async def subscriber(self):
+        conn = await asyncpg.connect(
+            host=self.host,
+            port=self.port,
+            user=self.user,
+            password=self.password,
+            database=self.dbname,
+            ssl=self.sslctx
+        )
+        await conn.add_listener(self.channel_name, self.notify)
+        while self.running:
+            await asyncio.sleep(1)  # keep the coroutine alive
+
+    async def set_update_callback(self, fn_name: Callable):
+        print("runtime is set update callback", fn_name)
+        self.callback = fn_name
+
+    async def update(self):
+        conn = await asyncpg.connect(
+                host=self.host,
+                port=self.port,
+                user=self.user,
+                password=self.password,
+                database=self.dbname,
+                ssl=self.sslctx
+        )
+        async with conn.transaction():
+            await conn.execute(
+                f"NOTIFY {self.channel_name},'casbin policy update at {time.time()}'"
+            )
+        await conn.close()
+        return True

examples/rbac_model.conf

@@ -0,0 +1,14 @@
+[request_definition]
+r = sub, obj, act
+
+[policy_definition]
+p = sub, obj, act
+
+[role_definition]
+g = _, _
+
+[policy_effect]
+e = some(where (p.eft == allow))
+
+[matchers]
+m = (p.sub == "*" || g(r.sub, p.sub)) && r.obj == p.obj && (p.act == "*" || r.act == p.act)

examples/rbac_policy.csv

@@ -0,0 +1,5 @@
+p,alice,data1,read
+p,bob,data2,write
+p,data2_admin,data2,read
+p,data2_admin,data2,write
+g,alice,data2_admin

test/test_async_postgres_watcher.py

@@ -0,0 +1,56 @@
+# Copyright 2024 The casbin Authors. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+from async_postgres_watcher import AsyncPostgresWatcher
+
+# Please set up yourself config
+HOST = "127.0.0.1"
+PORT = 5432
+USER = "postgres"
+PASSWORD = "123456"
+DBNAME = "postgres"
+
+
+class TestConfig(unittest.IsolatedAsyncioTestCase):
+
+    async def asyncSetUp(self):
+        self.pg_watcher = AsyncPostgresWatcher(
+            host=HOST,
+            port=PORT,
+            user=USER,
+            password=PASSWORD,
+            dbname=DBNAME
+        )
+
+    async def test_update_pg_watcher(self):
+        assert await self.pg_watcher.update() is True
+
+    def test_default_update_callback(self):
+        assert self.pg_watcher.callback is None
+
+    async def test_add_update_callback(self):
+        def _test_callback():
+            pass
+
+        await self.pg_watcher.set_update_callback(_test_callback)
+        assert self.pg_watcher.callback == _test_callback
+
+    async def asyncTearDown(self):
+        self.pg_watcher.running = False
+
+
+if __name__ == "__main__":
+    unittest.main()