Merge pull request #1 from PropelAuth/rbac

Add support for custom roles and permissions

propelauth_fastapi/__init__.py

@@ -1,10 +1,11 @@
 from collections import namedtuple
+from typing import List
 
 from fastapi import Depends, HTTPException
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 from propelauth_py import TokenVerificationMetadata, init_base_auth, Auth
-from propelauth_py.errors import ForbiddenException, UnauthorizedException, UnexpectedException
-from propelauth_py.user import User, UserRole
+from propelauth_py.errors import ForbiddenException, UnauthorizedException
+from propelauth_py.user import User
 
 _security = HTTPBearer(auto_error=False)
 
@@ -48,25 +49,70 @@ def __call__(self, credentials: HTTPAuthorizationCredentials = Depends(_security
 
 
 def _require_org_member_wrapper(auth: Auth, debug_mode: bool):
-    def require_org_member(user: User, required_org_id: str, minimum_required_role: UserRole = None):
+    def require_org_member(user: User, required_org_id: str):
         try:
-            return auth.validate_org_access_and_get_org(user, required_org_id, minimum_required_role)
+            return auth.validate_org_access_and_get_org(user, required_org_id)
         except ForbiddenException as e:
-            if debug_mode:
-                raise HTTPException(status_code=403, detail=e.message)
-            else:
-                raise HTTPException(status_code=403)
-        except UnexpectedException as e:
-            if debug_mode:
-                raise HTTPException(status_code=500, detail=e.message)
-            else:
-                raise HTTPException(status_code=500)
+            _handle_forbidden_exception(e, debug_mode)
 
     return require_org_member
 
 
+def _require_org_member_with_minimum_role_wrapper(auth: Auth, debug_mode: bool):
+    def require_org_member_with_minimum_role(user: User, required_org_id: str, minimum_required_role: str):
+        try:
+            return auth.validate_minimum_org_role_and_get_org(user, required_org_id, minimum_required_role)
+        except ForbiddenException as e:
+            _handle_forbidden_exception(e, debug_mode)
+
+    return require_org_member_with_minimum_role
+
+
+def _require_org_member_with_exact_role_wrapper(auth: Auth, debug_mode: bool):
+    def require_org_member_with_exact_role(user: User, required_org_id: str, role: str):
+        try:
+            return auth.validate_exact_org_role_and_get_org(user, required_org_id, role)
+        except ForbiddenException as e:
+            _handle_forbidden_exception(e, debug_mode)
+
+    return require_org_member_with_exact_role
+
+
+def _require_org_member_with_permission_wrapper(auth: Auth, debug_mode: bool):
+    def require_org_member_with_permission(user: User, required_org_id: str, permission: str):
+        try:
+            return auth.validate_permission_and_get_org(user, required_org_id, permission)
+        except ForbiddenException as e:
+            _handle_forbidden_exception(e, debug_mode)
+
+    return require_org_member_with_permission
+
+
+def _require_org_member_with_all_permissions_wrapper(auth: Auth, debug_mode: bool):
+    def require_org_member_with_all_permissions(user: User, required_org_id: str, permissions: List[str]):
+        try:
+            return auth.validate_all_permissions_and_get_org(user, required_org_id, permissions)
+        except ForbiddenException as e:
+            _handle_forbidden_exception(e, debug_mode)
+
+    return require_org_member_with_all_permissions
+
+
+def _handle_forbidden_exception(e: ForbiddenException, debug_mode: bool):
+    if debug_mode:
+        raise HTTPException(status_code=403, detail=e.message)
+    else:
+        raise HTTPException(status_code=403)
+
+
+
 Auth = namedtuple("Auth", [
-    "require_user", "optional_user", "require_org_member",
+    "require_user", "optional_user",
+    "require_org_member",
+    "require_org_member_with_minimum_role",
+    "require_org_member_with_exact_role",
+    "require_org_member_with_permission",
+    "require_org_member_with_all_permissions",
     "fetch_user_metadata_by_user_id", "fetch_user_metadata_by_email", "fetch_user_metadata_by_username",
     "fetch_batch_user_metadata_by_user_ids",
     "fetch_batch_user_metadata_by_emails",
@@ -90,6 +136,10 @@ def init_auth(auth_url: str, api_key: str, token_verification_metadata: TokenVer
         require_user=RequiredUserDependency(auth, debug_mode),
         optional_user=OptionalUserDependency(auth),
         require_org_member=_require_org_member_wrapper(auth, debug_mode),
+        require_org_member_with_minimum_role=_require_org_member_with_minimum_role_wrapper(auth, debug_mode),
+        require_org_member_with_exact_role=_require_org_member_with_exact_role_wrapper(auth, debug_mode),
+        require_org_member_with_permission=_require_org_member_with_permission_wrapper(auth, debug_mode),
+        require_org_member_with_all_permissions=_require_org_member_with_all_permissions_wrapper(auth, debug_mode),
         fetch_user_metadata_by_user_id=auth.fetch_user_metadata_by_user_id,
         fetch_user_metadata_by_email=auth.fetch_user_metadata_by_email,
         fetch_user_metadata_by_username=auth.fetch_user_metadata_by_username,

requirements.txt

@@ -1,4 +1,5 @@
 fastapi
-propelauth-py
+propelauth-py==3.0.0b7
 pytest
 requests-mock
+httpx

setup.py

@@ -6,7 +6,7 @@
 
 setup(
     name="propelauth-fastapi",
-    version="1.1.4",
+    version="2.0.0b7",
     description="A FastAPI library for managing authentication, backed by PropelAuth",
     long_description=README,
     long_description_content_type="text/markdown",

tests/auth_helpers.py

@@ -19,11 +19,14 @@ def random_user_id():
     return str(uuid4())
 
 
-def random_org(user_role_str):
+def random_org(user_role_str, permissions=None):
+    # represents the incoming JSON from the auth server
     return {
         "org_id": str(uuid4()),
         "org_name": str(uuid4()),
-        "user_role": user_role_str
+        "user_role": user_role_str,
+        "inherited_user_roles_plus_current_role": [user_role_str],
+        "user_permissions": [] if permissions is None else permissions,
     }
 
 

tests/test_require_org_member.py

@@ -6,21 +6,20 @@
 
 from tests.auth_helpers import create_access_token, orgs_to_org_id_map, random_org, random_user_id
 from tests.conftest import HTTP_BASE_AUTH_URL
-from propelauth_fastapi import UserRole
 
 ROUTE_NAME = "/require_org_member_route"
 
 
 def test_require_org_member_without_auth(app, auth, client, rsa_keys):
-    create_route_expecting_user_and_org(app, auth, None, None, None)
+    create_route_expecting_user_and_org(app, auth, None, None)
     org_id = str(uuid4())
 
     response = client.get(route_for(org_id))
     assert response.status_code == 401
 
 
 def test_require_org_member_with_auth_but_no_org_membership(app, auth, client, rsa_keys):
-    create_route_expecting_user_and_org(app, auth, None, None, None)
+    create_route_expecting_user_and_org(app, auth, None, None)
     org_id = str(uuid4())
 
     user_id = random_user_id()
@@ -29,12 +28,29 @@ def test_require_org_member_with_auth_but_no_org_membership(app, auth, client, r
     assert response.status_code == 403
 
 
+def test_require_org_member_with_auth_and_org_member_min_role(app, auth, client, rsa_keys):
+    user_id = random_user_id()
+    org = random_org("Owner")
+    org_id_to_org_member_info = orgs_to_org_id_map([org])
+
+    create_route_expecting_user_and_org_min_role(app, auth, user_id, org, "Owner")
+
+    access_token = create_access_token({
+        "user_id": user_id,
+        "org_id_to_org_member_info": org_id_to_org_member_info
+    }, rsa_keys.private_pem)
+
+    response = client.get(route_for(org["org_id"]), headers={"Authorization": "Bearer " + access_token})
+    assert response.status_code == 200
+    assert response.text == "ok"
+
+
 def test_require_org_member_with_auth_and_org_member(app, auth, client, rsa_keys):
     user_id = random_user_id()
     org = random_org("Owner")
     org_id_to_org_member_info = orgs_to_org_id_map([org])
 
-    create_route_expecting_user_and_org(app, auth, user_id, org, UserRole.Owner)
+    create_route_expecting_user_and_org_exact_role(app, auth, user_id, org, "Owner")
 
     access_token = create_access_token({
         "user_id": user_id,
@@ -52,7 +68,7 @@ def test_require_org_member_with_auth_but_wrong_org_id(app, auth, client, rsa_ke
     org_id_to_org_member_info = orgs_to_org_id_map([org])
     wrong_org_id = str(uuid4())
 
-    create_route_expecting_user_and_org(app, auth, user_id, org, UserRole.Owner)
+    create_route_expecting_user_and_org_exact_role(app, auth, user_id, org, "Owner")
 
     access_token = create_access_token({
         "user_id": user_id,
@@ -69,7 +85,7 @@ def test_require_org_member_with_auth_but_no_permission(app, auth, client, rsa_k
     org = random_org("Member")
     org_id_to_org_member_info = orgs_to_org_id_map([org])
 
-    create_route_expecting_user_and_org(app, auth, user_id, org, UserRole.Admin)
+    create_route_expecting_user_and_org_min_role(app, auth, user_id, org, "Admin")
 
     access_token = create_access_token({
         "user_id": user_id,
@@ -85,7 +101,7 @@ def test_require_org_member_with_auth_with_permission(app, auth, client, rsa_key
     org = random_org("Admin")
     org_id_to_org_member_info = orgs_to_org_id_map([org])
 
-    create_route_expecting_user_and_org(app, auth, user_id, org, UserRole.Admin)
+    create_route_expecting_user_and_org_min_role(app, auth, user_id, org, "Admin")
 
     access_token = create_access_token({
         "user_id": user_id,
@@ -97,8 +113,74 @@ def test_require_org_member_with_auth_with_permission(app, auth, client, rsa_key
     assert response.text == "ok"
 
 
+def test_require_org_member_with_auth_by_permission(app, auth, client, rsa_keys):
+    user_id = random_user_id()
+    org = random_org("Admin", ["permA"])
+    org_id_to_org_member_info = orgs_to_org_id_map([org])
+
+    create_route_expecting_user_and_org_by_permission(app, auth, user_id, org, "permA")
+
+    access_token = create_access_token({
+        "user_id": user_id,
+        "org_id_to_org_member_info": org_id_to_org_member_info
+    }, rsa_keys.private_pem)
+
+    response = client.get(route_for(org["org_id"]), headers={"Authorization": "Bearer " + access_token})
+    assert response.status_code == 200
+    assert response.text == "ok"
+
+
+def test_require_org_member_with_auth_by_permission_missing(app, auth, client, rsa_keys):
+    user_id = random_user_id()
+    org = random_org("Admin", ["permA"])
+    org_id_to_org_member_info = orgs_to_org_id_map([org])
+
+    create_route_expecting_user_and_org_by_permission(app, auth, user_id, org, "permB")
+
+    access_token = create_access_token({
+        "user_id": user_id,
+        "org_id_to_org_member_info": org_id_to_org_member_info
+    }, rsa_keys.private_pem)
+
+    response = client.get(route_for(org["org_id"]), headers={"Authorization": "Bearer " + access_token})
+    assert response.status_code == 403
+
+
+def test_require_org_member_with_auth_by_permissions(app, auth, client, rsa_keys):
+    user_id = random_user_id()
+    org = random_org("Admin", ["permA", "permB", "permC"])
+    org_id_to_org_member_info = orgs_to_org_id_map([org])
+
+    create_route_expecting_user_and_org_by_permissions(app, auth, user_id, org, ["permA", "permC"])
+
+    access_token = create_access_token({
+        "user_id": user_id,
+        "org_id_to_org_member_info": org_id_to_org_member_info
+    }, rsa_keys.private_pem)
+
+    response = client.get(route_for(org["org_id"]), headers={"Authorization": "Bearer " + access_token})
+    assert response.status_code == 200
+    assert response.text == "ok"
+
+
+def test_require_org_member_with_auth_by_permissions_missing(app, auth, client, rsa_keys):
+    user_id = random_user_id()
+    org = random_org("Admin", ["permA"])
+    org_id_to_org_member_info = orgs_to_org_id_map([org])
+
+    create_route_expecting_user_and_org_by_permissions(app, auth, user_id, org, ["permA", "permB"])
+
+    access_token = create_access_token({
+        "user_id": user_id,
+        "org_id_to_org_member_info": org_id_to_org_member_info
+    }, rsa_keys.private_pem)
+
+    response = client.get(route_for(org["org_id"]), headers={"Authorization": "Bearer " + access_token})
+    assert response.status_code == 403
+
+
 def test_require_org_member_with_bad_header(app, auth, client, rsa_keys):
-    create_route_expecting_user_and_org(app, auth, None, None, None)
+    create_route_expecting_user_and_org(app, auth, None, None)
 
     user_id = random_user_id()
     org = random_org("Admin")
@@ -114,7 +196,7 @@ def test_require_org_member_with_bad_header(app, auth, client, rsa_keys):
 
 
 def test_require_org_member_with_wrong_token(app, auth, client, rsa_keys):
-    create_route_expecting_user_and_org(app, auth, None, None, None)
+    create_route_expecting_user_and_org(app, auth, None, None)
     org_id = str(uuid4())
 
     response = client.get(route_for(org_id), headers={"Authorization": "Bearer whatisthis"})
@@ -126,7 +208,7 @@ def test_require_org_member_with_expired_token(app, auth, client, rsa_keys):
     org = random_org("Owner")
     org_id_to_org_member_info = orgs_to_org_id_map([org])
 
-    create_route_expecting_user_and_org(app, auth, user_id, org, UserRole.Owner)
+    create_route_expecting_user_and_org_exact_role(app, auth, user_id, org, "Owner")
 
     access_token = create_access_token({
         "user_id": user_id,
@@ -142,7 +224,7 @@ def test_require_user_with_bad_issuer(app, auth, client, rsa_keys):
     org = random_org("Owner")
     org_id_to_org_member_info = orgs_to_org_id_map([org])
 
-    create_route_expecting_user_and_org(app, auth, user_id, org, UserRole.Owner)
+    create_route_expecting_user_and_org_exact_role(app, auth, user_id, org, "Owner")
 
     access_token = create_access_token({
         "user_id": user_id,
@@ -153,14 +235,53 @@ def test_require_user_with_bad_issuer(app, auth, client, rsa_keys):
     assert response.status_code == 401
 
 
-def create_route_expecting_user_and_org(app, auth, user_id, org, user_role):
+def create_route_expecting_user_and_org(app, auth, user_id, org):
+    @app.get(ROUTE_NAME)
+    async def route(org_id, current_user=Depends(auth.require_user)):
+        current_org = auth.require_org_member(current_user, org_id)
+        assert current_user.user_id == user_id
+        assert current_org.org_id == org["org_id"]
+        assert current_org.org_name == org["org_name"]
+        return PlainTextResponse("ok")
+
+
+def create_route_expecting_user_and_org_min_role(app, auth, user_id, org, min_required_role):
+    @app.get(ROUTE_NAME)
+    async def route(org_id, current_user=Depends(auth.require_user)):
+        current_org = auth.require_org_member_with_minimum_role(current_user, org_id, min_required_role)
+        assert current_user.user_id == user_id
+        assert current_org.org_id == org["org_id"]
+        assert current_org.org_name == org["org_name"]
+        return PlainTextResponse("ok")
+
+
+def create_route_expecting_user_and_org_exact_role(app, auth, user_id, org, role):
+    @app.get(ROUTE_NAME)
+    async def route(org_id, current_user=Depends(auth.require_user)):
+        current_org = auth.require_org_member_with_exact_role(current_user, org_id, role)
+        assert current_user.user_id == user_id
+        assert current_org.org_id == org["org_id"]
+        assert current_org.org_name == org["org_name"]
+        return PlainTextResponse("ok")
+
+
+def create_route_expecting_user_and_org_by_permission(app, auth, user_id, org, permission):
+    @app.get(ROUTE_NAME)
+    async def route(org_id, current_user=Depends(auth.require_user)):
+        current_org = auth.require_org_member_with_permission(current_user, org_id, permission)
+        assert current_user.user_id == user_id
+        assert current_org.org_id == org["org_id"]
+        assert current_org.org_name == org["org_name"]
+        return PlainTextResponse("ok")
+
+
+def create_route_expecting_user_and_org_by_permissions(app, auth, user_id, org, permissions):
     @app.get(ROUTE_NAME)
     async def route(org_id, current_user=Depends(auth.require_user)):
-        current_org = auth.require_org_member(current_user, org_id, user_role)
+        current_org = auth.require_org_member_with_all_permissions(current_user, org_id, permissions)
         assert current_user.user_id == user_id
         assert current_org.org_id == org["org_id"]
         assert current_org.org_name == org["org_name"]
-        assert current_org.user_role == user_role
         return PlainTextResponse("ok")