Add regex for testing permission strings (#681)

Author: Dreamsorcerer

aiohttp_admin/__init__.py

@@ -1,3 +1,4 @@
+import re
 import secrets
 from typing import Optional
 
@@ -87,5 +88,21 @@ def value(r: web.RouteDef) -> tuple[str, str]:
 
     setup_routes(admin)
     setup_resources(admin, schema)
+
+    resource_patterns = []
+    for r, state in admin["state"]["resources"].items():
+        fields = state["fields"].keys()
+        resource_patterns.append(
+            r"(?#Resource name){r}"
+            r"(?#Optional field name)(\.({f}))?"
+            r"(?#Permission type)\.(view|edit|add|delete|\*)"
+            r"(?#No filters if negated)(?(2)$|"
+            r'(?#Optional filters)\|({f})=(?#JSON number or str)(\".*?\"|\d+))*'.format(
+                r=r, f="|".join(fields)))
+    p_re = (r"(?#Global admin permission)~?admin\.(view|edit|add|delete|\*)"
+            r"|"
+            r"(?#Resource permission)(~)?admin\.({})").format("|".join(resource_patterns))
+    admin["permission_re"] = re.compile(p_re)
+
     prefixed_subapp = app.add_subapp(path, admin)
     return admin

examples/permissions.py

@@ -1,7 +1,7 @@
 """Example to demonstrate usage of permissions.
 
 When running this file, admin will be accessible at /admin.
-Check below for valid usernames (and their respective permissions),
+Check near the bottom of the file for valid usernames (and their respective permissions),
 login will work with any password.
 """
 
@@ -46,37 +46,6 @@ async def create_app() -> web.Application:
     # Create some sample data
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
-    async with session.begin() as sess:
-        # Users with various permissions.
-        sess.add(User(username="admin", permissions=json.dumps(tuple(Permissions))))
-        sess.add(User(username="view", permissions=json.dumps((Permissions.view,))))
-        sess.add(User(username="add", permissions=json.dumps(
-            (Permissions.view, Permissions.add,))))
-        sess.add(User(username="edit", permissions=json.dumps(
-            (Permissions.view, Permissions.edit))))
-        sess.add(User(username="delete", permissions=json.dumps(
-            (Permissions.view, Permissions.delete))))
-        sess.add(User(username="simple", permissions=json.dumps(("admin.simple.*",))))
-        sess.add(User(username="mixed", permissions=json.dumps(
-            ("admin.simple.view", "admin.simple.edit", "admin.parent.view"))))
-        sess.add(User(username="negated", permissions=json.dumps(
-            ("admin.*", "~admin.parent.*", "~admin.simple.edit"))))
-        sess.add(User(username="field", permissions=json.dumps(
-            ("admin.*", "~admin.simple.optional_num.*"))))
-        sess.add(User(username="field_edit", permissions=json.dumps(
-            ("admin.*", "~admin.simple.optional_num.edit"))))
-        sess.add(User(username="filter", permissions=json.dumps(
-            ("admin.*", "admin.simple.*|num=5"))))
-        sess.add(User(username="filter_edit", permissions=json.dumps(
-            ("admin.*", "admin.simple.edit|num=5"))))
-        sess.add(User(username="filter_add", permissions=json.dumps(
-            ("admin.*", "admin.simple.add|num=5"))))
-        sess.add(User(username="filter_delete", permissions=json.dumps(
-            ("admin.*", "admin.simple.delete|num=5"))))
-        sess.add(User(username="filter_field", permissions=json.dumps(
-            ("admin.*", "admin.simple.optional_num.*|num=5"))))
-        sess.add(User(username="filter_field_edit", permissions=json.dumps(
-            ("admin.*", "admin.simple.optional_num.edit|num=5"))))
     async with session.begin() as sess:
         sess.add(Simple(num=5, value="first"))
         p = Simple(num=82, optional_num=12, value="with child")
@@ -104,7 +73,35 @@ async def create_app() -> web.Application:
             {"model": SAResource(engine, SimpleParent)}
         )
     }
-    aiohttp_admin.setup(app, schema)
+    admin = aiohttp_admin.setup(app, schema)
+
+    # Create users with various permissions.
+    async with session.begin() as sess:
+        sess.add(User(username="admin", permissions=json.dumps(tuple(Permissions))))
+        sess.add(User(username="view", permissions=json.dumps((Permissions.view,))))
+        sess.add(User(username="add", permissions=json.dumps(
+            (Permissions.view, Permissions.add,))))
+        sess.add(User(username="edit", permissions=json.dumps(
+            (Permissions.view, Permissions.edit))))
+        sess.add(User(username="delete", permissions=json.dumps(
+            (Permissions.view, Permissions.delete))))
+        users = {
+            "simple": ("admin.simple.*",),
+            "mixed": ("admin.simple.view", "admin.simple.edit", "admin.parent.view"),
+            "negated": ("admin.*", "~admin.parent.*", "~admin.simple.edit"),
+            "field": ("admin.*", "~admin.simple.optional_num.*"),
+            "field_edit": ("admin.*", "~admin.simple.optional_num.edit"),
+            "filter": ("admin.*", "admin.simple.*|num=5"),
+            "filter_edit": ("admin.*", "admin.simple.edit|num=5"),
+            "filter_add": ("admin.*", "admin.simple.add|num=5"),
+            "filter_delete": ("admin.*", "admin.simple.delete|num=5"),
+            "filter_field": ("admin.*", "admin.simple.optional_num.*|num=5"),
+            "filter_field_edit": ("admin.*", "admin.simple.optional_num.edit|num=5")
+        }
+        for name, permissions in users.items():
+            if any(admin["permission_re"].fullmatch(p) is None for p in permissions):
+                raise ValueError("Not a valid permission.")
+            sess.add(User(username=name, permissions=json.dumps(permissions)))
 
     return app
 

tests/test_admin.py

@@ -1,7 +1,10 @@
 from aiohttp import web
+from sqlalchemy.ext.asyncio import AsyncEngine
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 
 import aiohttp_admin
 from _auth import check_credentials
+from aiohttp_admin.backends.sqlalchemy import SAResource
 
 
 def test_path() -> None:
@@ -15,3 +18,38 @@ def test_path() -> None:
     admin = aiohttp_admin.setup(app, schema, path="/another/admin")
 
     assert str(admin.router["index"].url_for()) == "/another/admin"
+
+
+def test_re(base: type[DeclarativeBase], mock_engine: AsyncEngine) -> None:
+    class TestRE(base):  # type: ignore[misc,valid-type]
+        __tablename__ = "testre"
+        id: Mapped[int] = mapped_column(primary_key=True)
+        value: Mapped[str]
+
+    app = web.Application()
+    schema: aiohttp_admin.Schema = {"security": {"check_credentials": check_credentials},
+                                    "resources": ({"model": SAResource(mock_engine, TestRE)},)}
+    admin = aiohttp_admin.setup(app, schema)
+    r = admin["permission_re"]
+
+    assert r.fullmatch("admin.*")
+    assert r.fullmatch("admin.view")
+    assert r.fullmatch("~admin.edit")
+    assert r.fullmatch("admin.testre.*")
+    assert r.fullmatch("admin.testre.add")
+    assert r.fullmatch("admin.testre.id.*")
+    assert r.fullmatch("admin.testre.value.edit")
+    assert r.fullmatch("~admin.testre.id.edit")
+    assert r.fullmatch("admin.testre.edit|id=5")
+    assert r.fullmatch('admin.testre.add|id=1|value="4"|value="7"')
+    assert r.fullmatch('admin.testre.value.*|value="foo"')
+    assert r.fullmatch("admin.testre.value.delete|id=5|id=3")
+
+    assert r.fullmatch("testre.edit") is None
+    assert r.fullmatch("admin.create") is None
+    assert r.fullmatch("admin.nottest.*") is None
+    assert r.fullmatch("admin.*|id=1") is None
+    assert r.fullmatch("admin.testre.edit|other=5") is None
+    assert r.fullmatch("admin.testre.value.*|value=unquoted") is None
+    assert r.fullmatch("~admin.testre.edit|id=5") is None
+    assert r.fullmatch('~admin.testre.value.delete|value="1"') is None