Fix/rbac create sibling hardening

api/app/rbac_roles.py

@@ -6,12 +6,13 @@
     "custom",
 }
 
+DENIED_CREATOR_ROLES = {"viewer"}
+
 DB_ROLE_BY_RBAC_ROLE = {
     "viewer": "user",
     "editor": "user",
     "obs_manager": "sensor",
     "sensor": "sensor",
-    # Custom policies still require baseline schema/table permissions.
     "custom": "user",
 }
 
@@ -28,3 +29,9 @@ def validate_rbac_role(role: str) -> str:
 
 def get_db_role_for_rbac(role: str) -> str:
     return DB_ROLE_BY_RBAC_ROLE[validate_rbac_role(role)]
+
+
+def check_create_permission(role) -> bool:
+    if role is None:
+        return True
+    return role.lower() not in DENIED_CREATOR_ROLES

api/app/v1/endpoints/create/bulk_observation.py

@@ -15,7 +15,8 @@
 from app import AUTHORIZATION, POSTGRES_PORT_WRITE, VERSIONING
 from app.db.asyncpg_db import get_pool, get_pool_w
 from app.oauth import get_current_user
-from app.utils.utils import safe_parse_datetime
+from app.rbac_roles import check_create_permission
+from app.utils.utils import extract_iot_id, safe_parse_datetime
 from app.v1.endpoints.functions import set_role
 from asyncpg.exceptions import InsufficientPrivilegeError
 from asyncpg.types import Range
@@ -84,6 +85,18 @@ async def bulk_observations(
     current_user=user,
     pgpool=Depends(get_pool_w) if POSTGRES_PORT_WRITE else Depends(get_pool),
 ):
+    if current_user is not None:
+        user_role = current_user.get("role", "")
+        if not check_create_permission(user_role):
+            return JSONResponse(
+                status_code=status.HTTP_403_FORBIDDEN,
+                content={
+                    "code": 403,
+                    "type": "error",
+                    "message": "Insufficient privileges.",
+                },
+            )
+
     try:
         async with pgpool.acquire() as conn:
             async with conn.transaction():

api/app/v1/endpoints/create/data_array_observation.py

@@ -17,12 +17,16 @@
 
 from app import AUTHORIZATION, POSTGRES_PORT_WRITE, VERSIONING
 from app.db.asyncpg_db import get_pool, get_pool_w
+from app.rbac_roles import check_create_permission
+from app.v1.endpoints.functions import set_role
 from app.utils.utils import (
     build_self_link,
     check_iot_id_in_payload,
     check_missing_properties,
     handle_datetime_fields,
     handle_result_field,
+    build_self_link,
+    extract_iot_id,
 )
 from app.v1.endpoints.functions import set_role
 from asyncpg.exceptions import InsufficientPrivilegeError
@@ -99,6 +103,18 @@ async def data_array_observation(
     current_user=user,
     pool=Depends(get_pool_w) if POSTGRES_PORT_WRITE else Depends(get_pool),
 ):
+    if current_user is not None:
+        user_role = current_user.get("role", "")
+        if not check_create_permission(user_role):
+            return JSONResponse(
+                status_code=status.HTTP_403_FORBIDDEN,
+                content={
+                    "code": 403,
+                    "type": "error",
+                    "message": "Insufficient privileges.",
+                },
+            )
+
     try:
         response_urls = []
 

api/app/v1/endpoints/create/datastream.py

@@ -14,7 +14,8 @@
 
 from app import AUTHORIZATION, POSTGRES_PORT_WRITE, VERSIONING
 from app.db.asyncpg_db import get_pool, get_pool_w
-from app.utils.utils import require_json_content_type, validate_payload_keys
+from app.rbac_roles import check_create_permission
+from app.utils.utils import validate_payload_keys, require_json_content_type
 from app.v1.endpoints.functions import set_role
 from asyncpg.exceptions import InsufficientPrivilegeError
 from fastapi import APIRouter, Body, Depends, Header, Request, status
@@ -67,6 +68,8 @@
 if AUTHORIZATION:
     ALLOWED_KEYS.append("Network")
 
+DENIED_CREATOR_ROLES = ["viewer"]
+
 
 @v1.api_route(
     "/Datastreams",
@@ -83,6 +86,18 @@ async def create_datastream(
     current_user=user,
     pool=Depends(get_pool_w) if POSTGRES_PORT_WRITE else Depends(get_pool),
 ):
+    if current_user is not None:
+        user_role = current_user.get("role", "")
+        if not check_create_permission(user_role):
+            return JSONResponse(
+                status_code=status.HTTP_403_FORBIDDEN,
+                content={
+                    "code": 403,
+                    "type": "error",
+                    "message": "Insufficient privileges.",
+                },
+            )
+
     try:
         require_json_content_type(request)
 
@@ -91,6 +106,8 @@ async def create_datastream(
         async with pool.acquire() as connection:
             async with connection.transaction():
                 if current_user is not None:
+                    if current_user["role"] in DENIED_CREATOR_ROLES:
+                        raise InsufficientPrivilegeError
                     await set_role(connection, current_user)
 
                 commit_id = await set_commit(
@@ -159,6 +176,18 @@ async def create_datastream_for_thing(
     current_user=user,
     pool=Depends(get_pool_w) if POSTGRES_PORT_WRITE else Depends(get_pool),
 ):
+    if current_user is not None:
+        user_role = current_user.get("role", "")
+        if not check_create_permission(user_role):
+            return JSONResponse(
+                status_code=status.HTTP_403_FORBIDDEN,
+                content={
+                    "code": 403,
+                    "type": "error",
+                    "message": "Insufficient privileges.",
+                },
+            )
+
     try:
         require_json_content_type(request)
 
@@ -172,6 +201,8 @@ async def create_datastream_for_thing(
         async with pool.acquire() as connection:
             async with connection.transaction():
                 if current_user is not None:
+                    if current_user["role"] in DENIED_CREATOR_ROLES:
+                        raise InsufficientPrivilegeError
                     await set_role(connection, current_user)
 
                 commit_id = await set_commit(
@@ -240,6 +271,18 @@ async def create_datastream_for_sensor(
     current_user=user,
     pool=Depends(get_pool_w) if POSTGRES_PORT_WRITE else Depends(get_pool),
 ):
+    if current_user is not None:
+        user_role = current_user.get("role", "")
+        if not check_create_permission(user_role):
+            return JSONResponse(
+                status_code=status.HTTP_403_FORBIDDEN,
+                content={
+                    "code": 403,
+                    "type": "error",
+                    "message": "Insufficient privileges.",
+                },
+            )
+
     try:
         require_json_content_type(request)
 
@@ -253,6 +296,8 @@ async def create_datastream_for_sensor(
         async with pool.acquire() as connection:
             async with connection.transaction():
                 if current_user is not None:
+                    if current_user["role"] in DENIED_CREATOR_ROLES:
+                        raise InsufficientPrivilegeError
                     await set_role(connection, current_user)
 
                 commit_id = await set_commit(
@@ -324,19 +369,33 @@ async def create_datastream_for_observed_property(
     current_user=user,
     pool=Depends(get_pool_w) if POSTGRES_PORT_WRITE else Depends(get_pool),
 ):
+    if current_user is not None:
+        user_role = current_user.get("role", "")
+        if not check_create_permission(user_role):
+            return JSONResponse(
+                status_code=status.HTTP_403_FORBIDDEN,
+                content={
+                    "code": 403,
+                    "type": "error",
+                    "message": "Insufficient privileges.",
+                },
+            )
+
     try:
         require_json_content_type(request)
 
         if not observed_property_id:
             raise Exception("Observed Property ID is required.")
-
+ 
         payload["ObservedProperty"] = {"@iot.id": observed_property_id}
 
         validate_payload_keys(payload, ALLOWED_KEYS)
 
         async with pool.acquire() as connection:
             async with connection.transaction():
                 if current_user is not None:
+                    if current_user["role"] in DENIED_CREATOR_ROLES:
+                        raise InsufficientPrivilegeError
                     await set_role(connection, current_user)
 
                 commit_id = await set_commit(

api/app/v1/endpoints/create/feature_of_interest.py

@@ -14,11 +14,7 @@
 
 from app import AUTHORIZATION, POSTGRES_PORT_WRITE, VERSIONING
 from app.db.asyncpg_db import get_pool, get_pool_w
-from app.utils.utils import (
-    require_json_content_type,
-    validate_payload_keys,
-    validate_required_keys,
-)
+from app.utils.utils import validate_payload_keys, validate_required_keys, require_json_content_type
 from app.v1.endpoints.functions import set_role
 from asyncpg.exceptions import InsufficientPrivilegeError
 from fastapi import APIRouter, Body, Depends, Header, Request, status
@@ -56,6 +52,7 @@
 ]
 
 REQUIRED_KEYS = ["name", "encodingType", "feature"]
+DENIED_CREATOR_ROLES = ["viewer"]
 
 
 @v1.api_route(
@@ -73,6 +70,18 @@ async def create_feature_of_interest(
     current_user=user,
     pool=Depends(get_pool_w) if POSTGRES_PORT_WRITE else Depends(get_pool),
 ):
+    if current_user is not None:
+        user_role = current_user.get("role", "")
+        if not check_create_permission(user_role):
+            return JSONResponse(
+                status_code=status.HTTP_403_FORBIDDEN,
+                content={
+                    "code": 403,
+                    "type": "error",
+                    "message": "Insufficient privileges.",
+                },
+            )
+
     try:
         require_json_content_type(request)
 
@@ -82,6 +91,8 @@ async def create_feature_of_interest(
         async with pool.acquire() as connection:
             async with connection.transaction():
                 if current_user is not None:
+                    if current_user["role"] in DENIED_CREATOR_ROLES:
+                        raise InsufficientPrivilegeError
                     await set_role(connection, current_user)
 
                 commit_id = await set_commit(

api/app/v1/endpoints/create/historical_location.py

@@ -14,7 +14,8 @@
 
 from app import AUTHORIZATION, POSTGRES_PORT_WRITE, VERSIONING
 from app.db.asyncpg_db import get_pool, get_pool_w
-from app.utils.utils import require_json_content_type, validate_payload_keys
+from app.rbac_roles import check_create_permission
+from app.utils.utils import validate_payload_keys, require_json_content_type
 from app.v1.endpoints.functions import set_role
 from asyncpg.exceptions import InsufficientPrivilegeError
 from fastapi import APIRouter, Body, Depends, Header, Request, status
@@ -59,6 +60,18 @@ async def create_historical_location(
     current_user=user,
     pool=Depends(get_pool_w) if POSTGRES_PORT_WRITE else Depends(get_pool),
 ):
+    if current_user is not None:
+        user_role = current_user.get("role", "")
+        if not check_create_permission(user_role):
+            return JSONResponse(
+                status_code=status.HTTP_403_FORBIDDEN,
+                content={
+                    "code": 403,
+                    "type": "error",
+                    "message": "Insufficient privileges.",
+                },
+            )
+
     try:
         require_json_content_type(request)
 
@@ -68,7 +81,7 @@ async def create_historical_location(
             async with connection.transaction():
                 if current_user is not None:
                     await set_role(connection, current_user)
-
+ 
                 commit_id = await set_commit(
                     connection, commit_message, current_user
                 )
@@ -127,6 +140,18 @@ async def create_historical_location_for_thing(
     current_user=user,
     pool=Depends(get_pool_w) if POSTGRES_PORT_WRITE else Depends(get_pool),
 ):
+    if current_user is not None:
+        user_role = current_user.get("role", "")
+        if not check_create_permission(user_role):
+            return JSONResponse(
+                status_code=status.HTTP_403_FORBIDDEN,
+                content={
+                    "code": 403,
+                    "type": "error",
+                    "message": "Insufficient privileges.",
+                },
+            )
+
     try:
         require_json_content_type(request)
 

api/app/v1/endpoints/create/location.py

@@ -14,11 +14,8 @@
 
 from app import AUTHORIZATION, POSTGRES_PORT_WRITE, VERSIONING
 from app.db.asyncpg_db import get_pool, get_pool_w
-from app.utils.utils import (
-    require_json_content_type,
-    validate_payload_keys,
-    validate_required_keys,
-)
+from app.rbac_roles import check_create_permission
+from app.utils.utils import validate_payload_keys, validate_required_keys, require_json_content_type
 from app.v1.endpoints.functions import set_role
 from asyncpg.exceptions import InsufficientPrivilegeError
 from fastapi import APIRouter, Body, Depends, Header, Request, status
@@ -73,6 +70,18 @@ async def create_location(
     current_user=user,
     pool=Depends(get_pool_w) if POSTGRES_PORT_WRITE else Depends(get_pool),
 ):
+    if current_user is not None:
+        user_role = current_user.get("role", "")
+        if not check_create_permission(user_role):
+            return JSONResponse(
+                status_code=status.HTTP_403_FORBIDDEN,
+                content={
+                    "code": 403,
+                    "type": "error",
+                    "message": "Insufficient privileges.",
+                },
+            )
+
     try:
         require_json_content_type(request)
 
@@ -136,6 +145,18 @@ async def create_location_for_thing(
     current_user=user,
     pool=Depends(get_pool_w) if POSTGRES_PORT_WRITE else Depends(get_pool),
 ):
+    if current_user is not None:
+        user_role = current_user.get("role", "")
+        if not check_create_permission(user_role):
+            return JSONResponse(
+                status_code=status.HTTP_403_FORBIDDEN,
+                content={
+                    "code": 403,
+                    "type": "error",
+                    "message": "Insufficient privileges.",
+                },
+            )
+
     try:
         require_json_content_type(request)