Set Up Casbin RBAC Foundation with DB Table, Model, Policies, and Tests

backend/app/alembic/versions/b5055f0b4a4d_added_casbin_rule.py

@@ -0,0 +1,46 @@
+"""added casbin rule
+
+Revision ID: b5055f0b4a4d
+Revises: 8d7a05fd0ad4
+Create Date: 2025-04-09 13:45:43.511977
+
+"""
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+
+
+# revision identifiers, used by Alembic.
+revision = "b5055f0b4a4d"
+down_revision = "8d7a05fd0ad4"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "casbin_rule",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column(
+            "ptype", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False
+        ),
+        sa.Column("v0", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=True),
+        sa.Column("v1", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=True),
+        sa.Column("v2", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=True),
+        sa.Column("v3", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=True),
+        sa.Column("v4", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=True),
+        sa.Column("v5", sqlmodel.sql.sqltypes.AutoString(length=255), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_casbin_rule_ptype"), "casbin_rule", ["ptype"], unique=False
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(op.f("ix_casbin_rule_ptype"), table_name="casbin_rule")
+    op.drop_table("casbin_rule")
+    # ### end Alembic commands ###

backend/app/core/rbac/casbin.py

@@ -0,0 +1,16 @@
+import os
+from casbin import Enforcer
+from casbin_sqlalchemy_adapter import Adapter
+from app.core.db import engine
+from fastapi.concurrency import run_in_threadpool
+
+
+config_path = os.path.join(os.path.dirname(__file__), "rbac_model.conf")
+
+adapter = Adapter(engine)
+enforcer = Enforcer(config_path, adapter)
+enforcer.enable_auto_save(True)
+
+
+async def load_policy():
+    await run_in_threadpool(enforcer.load_policy)

backend/app/core/rbac/rbac_model.conf

@@ -0,0 +1,22 @@
+[request_definition]
+r = sub, org, proj, obj, act
+
+[policy_definition]
+p = role, obj, act
+
+[role_definition]
+g = _, _, _   # user → role → org
+g2 = _, _, _  # user → role → project
+
+[policy_effect]
+e = some(where (p.eft == allow))
+
+[matchers]
+# Matcher logic:
+# - If project ID is not present → check org-level role (g).
+# - If project ID is present:
+#   - Check if user has equivalent org-level role (e.g., org_admin → project_admin).
+#   - Otherwise, fall back to project-level role check (g2).
+# Final match only happens if object and action match as well.
+
+m = ((r.proj == "" && g(r.sub, p.role, r.org)) || (r.proj != "" && (g(r.sub, "org_admin", r.org) && p.role == "project_admin" || g(r.sub, "org_manager", r.org) && p.role == "project_manager" || g(r.sub, "org_reader", r.org) && p.role == "project_reader" || g2(r.sub, p.role, r.proj)))) && r.obj == p.obj && r.act == p.act

backend/app/core/rbac/rbac_policies.json

@@ -0,0 +1,34 @@
+{
+  "permissions": [
+    {
+      "role": "org_admin",
+      "resource": "org_data",
+      "actions": ["delete", "write", "read"]
+    },
+    {
+      "role": "org_manager",
+      "resource": "org_data",
+      "actions": ["write", "read"]
+    },
+    {
+      "role": "org_reader",
+      "resource": "org_data",
+      "actions": ["read"]
+    },
+    {
+      "role": "project_admin",
+      "resource": "project_data",
+      "actions": ["delete", "write", "read"]
+    },
+    {
+      "role": "project_manager",
+      "resource": "project_data",
+      "actions": ["write", "read"]
+    },
+    {
+      "role": "project_reader",
+      "resource": "project_data",
+      "actions": ["read"]
+    }
+  ]
+}

backend/app/core/rbac/update_casbin_policies.py

@@ -0,0 +1,71 @@
+import os
+import json
+import logging
+
+from sqlmodel import Session
+from sqlalchemy.sql import text
+
+from app.core.db import engine
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+policies_file_path = os.path.join(os.path.dirname(__file__), "rbac_policies.json")
+
+
+def update_policies(session: Session) -> None:
+    """
+    Update Casbin policies from the local JSON file.
+    This deletes all existing 'p' policies and inserts new ones.
+    """
+    try:
+        with open(policies_file_path) as f:
+            data = json.load(f)
+    except FileNotFoundError:
+        raise ValueError(f"Policy file not found: {policies_file_path}")
+
+    conn = session.connection()
+
+    try:
+        # Clear existing policies
+        logger.info("Deleting existing Casbin policies")
+        conn.execute(text("DELETE FROM casbin_rule WHERE ptype = 'p'"))
+
+        # Insert new policies
+        for policy in data.get("permissions", []):
+            role = policy.get("role")
+            resource = policy.get("resource")
+            actions = policy.get("actions")
+
+            if not role or not resource or not isinstance(actions, list):
+                raise ValueError(f"Invalid policy entry: {policy}")
+
+            for action in actions:
+                conn.execute(
+                    text(
+                        """
+                        INSERT INTO casbin_rule (ptype, v0, v1, v2)
+                        VALUES (:ptype, :v0, :v1, :v2)
+                    """
+                    ),
+                    {"ptype": "p", "v0": role, "v1": resource, "v2": action},
+                )
+
+        session.commit()
+        logger.info("Casbin policies updated successfully.")
+
+    except Exception as e:
+        logger.error(f"Error updating Casbin policies: {e}")
+        session.rollback()
+        raise
+
+
+def main() -> None:
+    logger.info("Starting Casbin policy update")
+    with Session(engine) as session:
+        update_policies(session)
+    logger.info("Casbin policy update finished")
+
+
+if __name__ == "__main__":
+    main()

backend/app/main.py

@@ -7,6 +7,8 @@
 from app.api.main import api_router
 from app.api.deps import http_exception_handler
 from app.core.config import settings
+from contextlib import asynccontextmanager
+from app.core.rbac.casbin import enforcer, load_policy
 
 
 def custom_generate_unique_id(route: APIRoute) -> str:
@@ -16,10 +18,20 @@ def custom_generate_unique_id(route: APIRoute) -> str:
 if settings.SENTRY_DSN and settings.ENVIRONMENT != "local":
     sentry_sdk.init(dsn=str(settings.SENTRY_DSN), enable_tracing=True)
 
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Load Casbin policy on startup
+    await load_policy()
+    app.state.enforcer = enforcer
+    yield
+
+
 app = FastAPI(
     title=settings.PROJECT_NAME,
     openapi_url=f"{settings.API_V1_STR}/openapi.json",
     generate_unique_id_function=custom_generate_unique_id,
+    lifespan=lifespan,
 )
 
 # Set all CORS enabled origins

backend/app/models/__init__.py

@@ -1,8 +1,8 @@
 from sqlmodel import SQLModel
 
 from .auth import Token, TokenPayload
+from .casbin_rule import CasbinRule
 from .document import Document
-
 from .message import Message
 
 from .project_user import (

backend/app/models/casbin_rule.py

@@ -0,0 +1,41 @@
+from typing import Optional
+from sqlmodel import SQLModel, Field
+
+
+class CasbinRule(SQLModel, table=True):
+    """
+    Represents Casbin policy rules for RBAC.
+
+    Policy type (`ptype`):
+    - "p"  -> policy rule (permissions)
+    - "g"  -> subject-role assignment (org-level)
+    - "g2" -> subject-role assignment (project-level)
+
+    Field meanings based on `ptype`:
+
+    For ptype = "p" (permissions):
+        - v0: role
+        - v1: resource (e.g., "org_data", "project_data")
+        - v2: action (e.g., "read", "write", "delete")
+
+    For ptype = "g" (org-level role assignment):
+        - v0: user_id
+        - v1: role (e.g., "org_admin", "org_writer", etc.)
+        - v2: org_id
+
+    For ptype = "g2" (project-level role assignment):
+        - v0: user_id
+        - v1: role (e.g., "project_reader", etc.)
+        - v2: project_id
+    """
+
+    __tablename__ = "casbin_rule"
+
+    id: Optional[int] = Field(default=None, primary_key=True)
+    ptype: str = Field(index=True, max_length=255)
+    v0: Optional[str] = Field(default=None, max_length=255)
+    v1: Optional[str] = Field(default=None, max_length=255)
+    v2: Optional[str] = Field(default=None, max_length=255)
+    v3: Optional[str] = Field(default=None, max_length=255)
+    v4: Optional[str] = Field(default=None, max_length=255)
+    v5: Optional[str] = Field(default=None, max_length=255)

backend/app/tests/conftest.py

@@ -6,6 +6,7 @@
 
 from app.core.config import settings
 from app.core.db import engine, init_db
+from app.core.rbac.update_casbin_policies import update_policies
 from app.main import app
 from app.models import (
     APIKey,
@@ -22,6 +23,7 @@
 def db() -> Generator[Session, None, None]:
     with Session(engine) as session:
         init_db(session)
+        update_policies(session)
         yield session
         # Delete data in reverse dependency order
         session.execute(delete(ProjectUser))  # Many-to-many relationship