Merge pull request #63 from minvws/feat/possible-scopes

feat: allowed scopes as part of application config

Author: fouadsfarijlani

app.conf.example

@@ -1,6 +1,8 @@
 [app]
 # Loglevel can be one of: debug, info, warning, error, critical
 loglevel=debug
+# list of possible scopes according to the NVI 
+scopes = nvi:create nvi:read nvi:update nvi:delete nvi:localize 
 
 [database]
 # Dsn for database connection

app/config.py

@@ -2,9 +2,9 @@
 import logging
 import os
 from enum import Enum
-from typing import Any
+from typing import Any, List
 
-from pydantic import BaseModel, Field, ValidationError
+from pydantic import BaseModel, Field, ValidationError, field_validator
 
 logger = logging.getLogger(__name__)
 
@@ -24,6 +24,16 @@ class LogLevel(str, Enum):
 
 class ConfigApp(BaseModel):
     loglevel: LogLevel = Field(default=LogLevel.info)
+    scopes: List[str] = Field(default=[])
+
+    @field_validator("scopes", mode="before")
+    @classmethod
+    def validate_scopes(cls, value: Any) -> List[str]:
+        if not isinstance(value, str):
+            raise ValueError("only space separated str are allowed. Check config file..")
+
+        trimmed_value = value.lstrip().rstrip()
+        return trimmed_value.split(" ")
 
 
 class ConfigDatabase(BaseModel):

app/container.py

@@ -17,7 +17,7 @@ def container_config(binder: inject.Binder) -> None:
     db = Database(config_database=config.database)
     binder.bind(Database, db)
 
-    organization_service = OrganizationService(db)
+    organization_service = OrganizationService(db, config.app.scopes)
     binder.bind(OrganizationService, organization_service)
 
     client_service = ClientService(db, organization_service)

app/scope_utils.py

@@ -1,3 +1,6 @@
+from typing import List
+
+
 def parse(value: str | None) -> set[str]:
     if not value:
         return set()
@@ -6,3 +9,9 @@ def parse(value: str | None) -> set[str]:
 
 def is_subset(child: str | None, parent: str | None) -> bool:
     return parse(child).issubset(parse(parent))
+
+
+def check_allowed(allowed: List[str], requested: str) -> bool:
+    sanitized = requested.lstrip().rstrip().split(" ")
+
+    return set(sanitized).issubset(set(allowed))

app/services/exceptions.py

@@ -1,3 +1,8 @@
 class ScopesNotGrantedError(Exception):
     def __init__(self, ungranted: set[str]) -> None:
         super().__init__(f"Scopes not granted by the organization: {', '.join(sorted(ungranted))}")
+
+
+class ScopeNotAllowedError(Exception):
+    def __init__(self, scope: str) -> None:
+        super().__init__(f"Scope {scope} is not allowed")

app/services/organization.py

@@ -1,25 +1,31 @@
 from datetime import datetime
-from typing import List
+from typing import Any, List
 from uuid import UUID
 
 from app import scope_utils
 from app.db.db import Database
 from app.db.models.organization import OrganizationEntity
 from app.db.repository.organization import OrganizationRepository
 from app.models.ura import UraNumber
-from app.services.exceptions import ScopesNotGrantedError
+from app.services.exceptions import ScopeNotAllowedError, ScopesNotGrantedError
 
 
 class OrganizationService:
-    def __init__(self, db: Database) -> None:
+    def __init__(self, db: Database, allowed_scopes: List[str]) -> None:
         self.db = db
+        self.allowed_scopes = allowed_scopes
 
     def create_one(
         self,
         register_id: UraNumber,
         name: str,
         scopes: str | None = None,
     ) -> OrganizationEntity:
+        if scopes:
+            scope_allowed = scope_utils.check_allowed(self.allowed_scopes, scopes)
+            if not scope_allowed:
+                raise ScopeNotAllowedError(scopes)
+
         with self.db.get_db_session() as session:
             repo = session.get_repository(OrganizationRepository)
             entity = OrganizationEntity(register_id=register_id, name=name, scopes=scopes)
@@ -45,10 +51,21 @@ def get_many(
         with self.db.get_db_session() as session:
             repo = session.get_repository(OrganizationRepository)
             return list(
-                repo.get_many(register_id=register_id, name=name, scopes=scopes, include_deleted=include_deleted)
+                repo.get_many(
+                    register_id=register_id,
+                    name=name,
+                    scopes=scopes,
+                    include_deleted=include_deleted,
+                )
             )
 
-    def update_one(self, id: UUID, **kwargs: object) -> OrganizationEntity | None:
+    def update_one(self, id: UUID, **kwargs: Any) -> OrganizationEntity | None:
+        if "scopes" in kwargs:
+            scopes: str = kwargs["scopes"]
+            valid_scope = scope_utils.check_allowed(self.allowed_scopes, scopes)
+            if not valid_scope:
+                raise ScopeNotAllowedError(scopes)
+
         with self.db.get_db_session() as session:
             repo = session.get_repository(OrganizationRepository)
             return repo.update(id, **kwargs)

tests/conftest.py

@@ -52,7 +52,7 @@ def client_repository(database: Database) -> ClientRepository:
 
 @pytest.fixture()
 def organization_service(database: Database) -> OrganizationService:
-    return OrganizationService(database)
+    return OrganizationService(database, allowed_scopes=["read", "write", "delete"])
 
 
 @pytest.fixture()
@@ -66,7 +66,9 @@ def organization_entity() -> OrganizationEntity:
 
 
 @pytest.fixture()
-def persisted_organization(organization_service: OrganizationService) -> OrganizationEntity:
+def persisted_organization(
+    organization_service: OrganizationService,
+) -> OrganizationEntity:
     return organization_service.create_one(register_id=TEST_REGISTER_ID, name=TEST_ORG_NAME)