IFC-739 Remove references to account role attribute (#4990)

backend/infrahub/api/artifact.py

@@ -1,18 +1,25 @@
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from fastapi import APIRouter, Body, Depends, Request, Response
 from pydantic import BaseModel, Field
 
 from infrahub.api.dependencies import BranchParams, get_branch_params, get_current_user, get_db
 from infrahub.core import registry
-from infrahub.core.constants import InfrahubKind
+from infrahub.core.account import ObjectPermission
+from infrahub.core.constants import InfrahubKind, PermissionAction
 from infrahub.core.protocols import CoreArtifactDefinition
 from infrahub.database import InfrahubDatabase  # noqa: TCH001
-from infrahub.exceptions import NodeNotFoundError
+from infrahub.exceptions import NodeNotFoundError, PermissionDeniedError
 from infrahub.git.models import RequestArtifactDefinitionGenerate
 from infrahub.log import get_logger
+from infrahub.permissions.constants import PermissionDecisionFlag
 from infrahub.workflows.catalogue import REQUEST_ARTIFACT_DEFINITION_GENERATE
 
+if TYPE_CHECKING:
+    from infrahub.auth import AccountSession
+
 log = get_logger()
 router = APIRouter(prefix="/artifact")
 
@@ -54,8 +61,25 @@ async def generate_artifact(
     ),
     db: InfrahubDatabase = Depends(get_db),
     branch_params: BranchParams = Depends(get_branch_params),
-    _: str = Depends(get_current_user),
+    account_session: AccountSession = Depends(get_current_user),
 ) -> None:
+    permission_decision = (
+        PermissionDecisionFlag.ALLOW_DEFAULT
+        if branch_params.branch.name == registry.default_branch
+        else PermissionDecisionFlag.ALLOW_OTHER
+    )
+    for permission in [
+        ObjectPermission(namespace="Core", name="Artifact", action=action.value, decision=permission_decision)
+        for action in (PermissionAction.CREATE, PermissionAction.UPDATE)
+    ]:
+        has_permission = False
+        for permission_backend in registry.permission_backends:
+            has_permission = await permission_backend.has_permission(
+                db=db, account_session=account_session, permission=permission, branch=branch_params.branch
+            )
+        if not has_permission:
+            raise PermissionDeniedError(f"You do not have the following permission: {permission}")
+
     # Verify that the artifact definition exists for the requested branch
     artifact_definition = await registry.manager.get_one_by_id_or_default_filter(
         db=db,

backend/infrahub/api/dependencies.py

@@ -12,7 +12,7 @@
 from infrahub.core.registry import registry
 from infrahub.core.timestamp import Timestamp
 from infrahub.database import InfrahubDatabase  # noqa: TCH001
-from infrahub.exceptions import AuthorizationError, PermissionDeniedError
+from infrahub.exceptions import AuthorizationError
 
 if TYPE_CHECKING:
     from neo4j import AsyncSession
@@ -112,13 +112,11 @@ async def get_current_user(
 
     account_session = await authentication_token(db=db, jwt_token=jwt_token, api_key=api_key)
 
-    if account_session.authenticated or request.url.path.startswith("/graphql"):
+    if (
+        account_session.authenticated
+        or request.url.path.startswith("/graphql")
+        or (config.SETTINGS.main.allow_anonymous_access and request.method.lower() in ["get", "options"])
+    ):
         return account_session
 
-    if config.SETTINGS.main.allow_anonymous_access and request.method.lower() in ["get", "options"]:
-        return account_session
-
-    if request.method.lower() == "post" and account_session.read_only and account_session.authenticated:
-        raise PermissionDeniedError("You are not allowed to perform this operation")
-
     raise AuthorizationError("Authentication is required")

backend/infrahub/auth.py

@@ -3,7 +3,7 @@
 import uuid
 from datetime import datetime, timedelta, timezone
 from enum import Enum
-from typing import TYPE_CHECKING, Callable, Optional
+from typing import TYPE_CHECKING, Callable
 
 import bcrypt
 import jwt
@@ -34,14 +34,9 @@ class AuthType(str, Enum):
 class AccountSession(BaseModel):
     authenticated: bool = True
     account_id: str
-    session_id: Optional[str] = None
-    role: str = "read-only"
+    session_id: str | None = None
     auth_type: AuthType
 
-    @property
-    def read_only(self) -> bool:
-        return self.role == "read-only"
-
     @property
     def authenticated_by_jwt(self) -> bool:
         return self.auth_type == AuthType.JWT
@@ -54,7 +49,7 @@ async def validate_active_account(db: InfrahubDatabase, account_id: str) -> None
 
 
 async def authenticate_with_password(
-    db: InfrahubDatabase, credentials: models.PasswordCredential, branch: Optional[str] = None
+    db: InfrahubDatabase, credentials: models.PasswordCredential, branch: str | None = None
 ) -> models.UserToken:
     selected_branch = await registry.get_branch(db=db, branch=branch)
 
@@ -86,10 +81,8 @@ async def authenticate_with_password(
     now = datetime.now(tz=timezone.utc)
     refresh_expires = now + timedelta(seconds=config.SETTINGS.security.refresh_token_lifetime)
 
-    # The read-only account role is deprecated and will only be used for anonymous access
-    role = "read-write" if account.role.value.value == "read-only" else account.role.value.value
     session_id = await create_db_refresh_token(db=db, account_id=account.id, expiration=refresh_expires)
-    access_token = generate_access_token(account_id=account.id, role=role, session_id=session_id)
+    access_token = generate_access_token(account_id=account.id, session_id=session_id)
     refresh_token = generate_refresh_token(account_id=account.id, session_id=session_id, expiration=refresh_expires)
 
     return models.UserToken(access_token=access_token, refresh_token=refresh_token)
@@ -111,7 +104,7 @@ async def create_fresh_access_token(
     if not refresh_token:
         raise AuthorizationError("The provided refresh token has been invalidated in the database")
 
-    account: Optional[CoreGenericAccount] = await NodeManager.get_one(id=refresh_data.account_id, db=db)
+    account: CoreGenericAccount | None = await NodeManager.get_one(id=refresh_data.account_id, db=db)
     if not account:
         raise NodeNotFoundError(
             branch_name=selected_branch.name,
@@ -120,9 +113,7 @@ async def create_fresh_access_token(
             message="That login user doesn't exist in the system",
         )
 
-    access_token = generate_access_token(
-        account_id=account.id, role=account.role.value.value, session_id=refresh_data.session_id
-    )
+    access_token = generate_access_token(account_id=account.id, session_id=refresh_data.session_id)
 
     return models.AccessTokenResponse(access_token=access_token)
 
@@ -132,7 +123,7 @@ async def signin_sso_account(db: InfrahubDatabase, account_name: str, sso_groups
 
     if not account:
         account = await Node.init(db=db, schema=InfrahubKind.ACCOUNT)
-        await account.new(db=db, name=account_name, account_type="User", role="admin", password=str(uuid.uuid4()))
+        await account.new(db=db, name=account_name, account_type="User", password=str(uuid.uuid4()))
         await account.save(db=db)
 
     if sso_groups:
@@ -151,12 +142,12 @@ async def signin_sso_account(db: InfrahubDatabase, account_name: str, sso_groups
     now = datetime.now(tz=timezone.utc)
     refresh_expires = now + timedelta(seconds=config.SETTINGS.security.refresh_token_lifetime)
     session_id = await create_db_refresh_token(db=db, account_id=account.id, expiration=refresh_expires)
-    access_token = generate_access_token(account_id=account.id, role=account.role.value.value, session_id=session_id)
+    access_token = generate_access_token(account_id=account.id, session_id=session_id)
     refresh_token = generate_refresh_token(account_id=account.id, session_id=session_id, expiration=refresh_expires)
     return models.UserToken(access_token=access_token, refresh_token=refresh_token)
 
 
-def generate_access_token(account_id: str, role: str, session_id: uuid.UUID) -> str:
+def generate_access_token(account_id: str, session_id: uuid.UUID) -> str:
     now = datetime.now(tz=timezone.utc)
 
     access_expires = now + timedelta(seconds=config.SETTINGS.security.access_token_lifetime)
@@ -168,7 +159,6 @@ def generate_access_token(account_id: str, role: str, session_id: uuid.UUID) ->
         "fresh": False,
         "type": "access",
         "session_id": str(session_id),
-        "user_claims": {"role": role},
     }
     access_token = jwt.encode(access_data, config.SETTINGS.security.secret_key, algorithm="HS256")
     return access_token
@@ -191,7 +181,7 @@ def generate_refresh_token(account_id: str, session_id: uuid.UUID, expiration: d
 
 
 async def authentication_token(
-    db: InfrahubDatabase, jwt_token: Optional[str] = None, api_key: Optional[str] = None
+    db: InfrahubDatabase, jwt_token: str | None = None, api_key: str | None = None
 ) -> AccountSession:
     if api_key:
         return await validate_api_key(db=db, token=api_key)
@@ -205,15 +195,14 @@ async def validate_jwt_access_token(token: str) -> AccountSession:
     try:
         payload = jwt.decode(token, config.SETTINGS.security.secret_key, algorithms=["HS256"])
         account_id = payload["sub"]
-        role = payload["user_claims"]["role"]
         session_id = payload["session_id"]
     except jwt.ExpiredSignatureError:
         raise AuthorizationError("Expired Signature") from None
     except Exception:
         raise AuthorizationError("Invalid token") from None
 
     if payload["type"] == "access":
-        return AccountSession(account_id=account_id, role=role, session_id=session_id, auth_type=AuthType.JWT)
+        return AccountSession(account_id=account_id, session_id=session_id, auth_type=AuthType.JWT)
 
     raise AuthorizationError("Invalid token, current token is not an access token")
 
@@ -237,27 +226,16 @@ async def validate_jwt_refresh_token(db: InfrahubDatabase, token: str) -> models
 
 
 async def validate_api_key(db: InfrahubDatabase, token: str) -> AccountSession:
-    account_id, role = await validate_token(token=token, db=db)
+    account_id = await validate_token(token=token, db=db)
     if not account_id:
         raise AuthorizationError("Invalid token")
 
     await validate_active_account(db=db, account_id=str(account_id))
 
-    # The read-only account role is deprecated and will only be used for anonymous access
-    role = "read-write" if role == "read-only" else role
-
-    return AccountSession(account_id=account_id, role=role, auth_type=AuthType.API)
-
-
-def _validate_is_admin(account_session: AccountSession) -> None:
-    if account_session.role != "admin":
-        raise PermissionError("You are not authorized to perform this operation")
+    return AccountSession(account_id=account_id, auth_type=AuthType.API)
 
 
 def _validate_update_account(account_session: AccountSession, node_id: str, fields: list[str]) -> None:
-    if account_session.role == "admin":
-        return
-
     if account_session.account_id != node_id:
         # A regular account is not allowed to modify another account
         raise PermissionError("You are not allowed to modify this account")
@@ -268,16 +246,6 @@ def _validate_update_account(account_session: AccountSession, node_id: str, fiel
             raise PermissionError(f"You are not allowed to modify '{field}'")
 
 
-def validate_mutation_permissions(operation: str, account_session: AccountSession) -> None:
-    validation_map: dict[str, Callable[[AccountSession], None]] = {
-        f"{InfrahubKind.ACCOUNT}Create": _validate_is_admin,
-        f"{InfrahubKind.ACCOUNT}Delete": _validate_is_admin,
-        f"{InfrahubKind.ACCOUNT}Upsert": _validate_is_admin,
-    }
-    if validator := validation_map.get(operation):
-        validator(account_session)
-
-
 def validate_mutation_permissions_update_node(
     operation: str, node_id: str, account_session: AccountSession, fields: list[str]
 ) -> None:

backend/infrahub/core/account.py

@@ -554,9 +554,7 @@ async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None:
         self.params.update(token_params)
 
         account_filter_perms, account_params = self.branch.get_query_filter_relationships(
-            rel_labels=["r31", "r32", "r41", "r42", "r5", "r6", "r7", "r8"],
-            at=self.at,
-            include_outside_parentheses=True,
+            rel_labels=["r31", "r41", "r5", "r6"], at=self.at, include_outside_parentheses=True
         )
         self.params.update(account_params)
 
@@ -568,7 +566,6 @@ async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None:
         WHERE %s
         WITH at
         MATCH (at)-[r31]-(:Relationship)-[r41]-(acc:CoreGenericAccount)-[r5:HAS_ATTRIBUTE]-(an:Attribute {name: "name"})-[r6:HAS_VALUE]-(av:AttributeValue)
-        MATCH (at)-[r32]-(:Relationship)-[r42]-(acc:CoreGenericAccount)-[r7:HAS_ATTRIBUTE]-(ar:Attribute {name: "role"})-[r8:HAS_VALUE]-(avr:AttributeValue)
         WHERE %s
         """ % (
             "\n AND ".join(token_filter_perms),
@@ -577,7 +574,7 @@ async def query_init(self, db: InfrahubDatabase, **kwargs: Any) -> None:
 
         self.add_to_query(query)
 
-        self.return_labels = ["at", "av", "avr", "acc"]
+        self.return_labels = ["at", "av", "acc"]
 
     def get_account_name(self) -> Optional[str]:
         """Return the account name that matched the query or None."""
@@ -593,18 +590,9 @@ def get_account_id(self) -> Optional[str]:
 
         return None
 
-    def get_account_role(self) -> str:
-        """Return the account role that matched the query or a None."""
-        if result := self.get_result():
-            return result.get("avr").get("value")
-
-        return "read-only"
-
 
-async def validate_token(
-    token: str, db: InfrahubDatabase, branch: Optional[Union[Branch, str]] = None
-) -> tuple[Optional[str], str]:
+async def validate_token(token: str, db: InfrahubDatabase, branch: Optional[Union[Branch, str]] = None) -> str | None:
     branch = await registry.get_branch(db=db, branch=branch)
     query = await AccountTokenValidateQuery.init(db=db, branch=branch, token=token)
     await query.execute(db=db)
-    return query.get_account_id(), query.get_account_role()
+    return query.get_account_id()

backend/infrahub/core/constants/__init__.py

@@ -77,12 +77,6 @@ class PermissionDecision(InfrahubNumberEnum):
     ALLOW_ALL = 6
 
 
-class AccountRole(InfrahubStringEnum):
-    ADMIN = "admin"
-    READ_ONLY = "read-only"
-    READ_WRITE = "read-write"
-
-
 class AccountType(InfrahubStringEnum):
     USER = "User"
     SCRIPT = "Script"

backend/infrahub/core/initialization.py

@@ -1,5 +1,4 @@
 import importlib
-from typing import Optional
 from uuid import uuid4
 
 from infrahub import config, lock
@@ -8,7 +7,6 @@
 from infrahub.core.constants import (
     DEFAULT_IP_NAMESPACE,
     GLOBAL_BRANCH_NAME,
-    AccountRole,
     GlobalPermissions,
     InfrahubKind,
     PermissionAction,
@@ -54,7 +52,7 @@ async def get_root_node(db: InfrahubDatabase, initialize: bool = False) -> Root:
     return roots[0]
 
 
-async def get_default_ipnamespace(db: InfrahubDatabase) -> Optional[Node]:
+async def get_default_ipnamespace(db: InfrahubDatabase) -> Node | None:
     if not registry.schema._branches or not registry.schema.has(name=InfrahubKind.NAMESPACE):
         return None
 
@@ -232,7 +230,7 @@ async def create_global_branch(db: InfrahubDatabase) -> Branch:
 
 
 async def create_branch(
-    branch_name: str, db: InfrahubDatabase, description: str = "", isolated: bool = True, at: Optional[str] = None
+    branch_name: str, db: InfrahubDatabase, description: str = "", isolated: bool = True, at: str | None = None
 ) -> Branch:
     """Create a new Branch, currently all the branches are based on Main
 
@@ -266,13 +264,12 @@ async def create_branch(
 async def create_account(
     db: InfrahubDatabase,
     name: str = "admin",
-    role: str = "admin",
-    password: Optional[str] = None,
-    token_value: Optional[str] = None,
+    password: str | None = None,
+    token_value: str | None = None,
 ) -> CoreAccount:
     token_schema = db.schema.get_node_schema(name=InfrahubKind.ACCOUNTTOKEN)
     obj = await Node.init(db=db, schema=CoreAccount)
-    await obj.new(db=db, name=name, account_type="User", role=role, password=password)
+    await obj.new(db=db, name=name, account_type="User", password=password)
     await obj.save(db=db)
     log.info(f"Created Account: {name}", account_name=name)
 
@@ -501,11 +498,7 @@ async def first_time_initialization(db: InfrahubDatabase) -> None:
 
         admin_accounts.append(
             await create_account(
-                db=db,
-                name="agent",
-                password=password,
-                role=AccountRole.READ_WRITE.value,
-                token_value=config.SETTINGS.initial.agent_token,
+                db=db, name="agent", password=password, token_value=config.SETTINGS.initial.agent_token
             )
         )
 

backend/infrahub/core/protocols.py

@@ -102,7 +102,6 @@ class CoreGenericAccount(CoreNode):
     label: StringOptional
     description: StringOptional
     account_type: Enum
-    role: Enum
     status: Dropdown
     tokens: RelationshipManager