fixed

horao/__init__.py

@@ -9,6 +9,7 @@
 """
 import logging
 import os
+from typing import Optional
 
 from opentelemetry import metrics, trace  # type: ignore
 from opentelemetry.instrumentation.logging import LoggingInstrumentor  # type: ignore
@@ -121,7 +122,7 @@ async def docs(request):
     return HTMLResponse(html)
 
 
-def init(authorization: AuthenticationBackend = None) -> Starlette:
+def init(authorization: Optional[AuthenticationBackend] = None) -> Starlette:
     """
     Initialize the API
     authorization: optional authorization backend to overwrite default behavior (useful for testing)

horao/api/synchronization.py

@@ -8,10 +8,13 @@
 from starlette.requests import Request
 from starlette.responses import JSONResponse
 
+from horao.auth.permissions import Namespace, Permission
+from horao.auth.validate import permission_required
 from horao.persistance import HoraoDecoder, init_session
 
 
-@requires("authenticated_peer")
+@requires("authenticated")
+@permission_required(Namespace.Peer, Permission.Write)
 async def synchronize(request: Request) -> JSONResponse:
     """
     responses:

horao/auth/__init__.py

@@ -5,5 +5,7 @@
 only contains authorization. The RBAC module is used to define the roles and permissions of the users. There are
 various implementations that can be used, but some are only meant for development purpose.
 """
-from horao.auth.basic import BasicAuthBackend
-from horao.auth.multi import MultiAuthBackend, Peer
+from __future__ import annotations
+
+from horao.auth.multi import MultiAuthBackend
+from horao.auth.roles import Peer

horao/auth/error.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+from typing import Callable, TypeVar
+
+RT = TypeVar("RT")
+
+
+class UnauthorizedError(RuntimeError):
+    """We raise this exception when a user tries to access a resource without the proper permissions."""
+
+    def __init__(self, function: Callable[..., RT], *args: str, **kwargs: int):
+        super().__init__("unauthorized access to resource")
+        self.function = function
+        self.args = args
+        self.kwargs = kwargs
+
+    def __str__(self):
+        return (
+            f"UnauthorizedError from [{self.function}] - ({self.args} ; {self.kwargs})"
+        )

horao/auth/multi.py

@@ -17,35 +17,7 @@
 )
 from starlette.requests import HTTPConnection
 
-
-class Peer(BaseUser):
-    def __init__(self, identity: str, token: str, payload, origin: str) -> None:
-        self.id = identity
-        self.token = token
-        self.payload = payload
-        self.origin = origin
-
-    @property
-    def is_authenticated(self) -> bool:
-        return True
-
-    @property
-    def display_name(self) -> str:
-        return self.origin
-
-    @property
-    def identity(self) -> str:
-        return self.id
-
-    def is_true(self) -> bool:
-        """
-        Check if the identity matches the origin.
-        :return: bool
-        """
-        return self.identity == self.origin
-
-    def __str__(self) -> str:
-        return f"{self.origin} -> {self.identity}"
+from horao.auth.roles import Peer
 
 
 class MultiAuthBackend(AuthenticationBackend):
@@ -54,20 +26,21 @@ class MultiAuthBackend(AuthenticationBackend):
     def digest_authentication(
         self, conn: HTTPConnection, token: str
     ) -> Union[None, Tuple[AuthCredentials, BaseUser]]:
+        host = conn.client.host  # type: ignore
         peer_match_source = False
         for peer in os.getenv("PEERS").split(","):  # type: ignore
-            if peer in conn.client.host:
+            if peer in host:
                 self.logger.debug(f"Peer {peer} is trying to authenticate")
                 peer_match_source = True
         if not peer_match_source and os.getenv("PEER_STRICT", "True") == "True":
-            raise AuthenticationError(f"access not allowed for {conn.client.host}")
+            raise AuthenticationError(f"access not allowed for {host}")
         payload = jwt.decode(token, os.getenv("PEER_SECRET"), algorithms=["HS256"])  # type: ignore
         self.logger.debug(f"valid token for {payload['peer']}")
-        return AuthCredentials(["authenticated_peer"]), Peer(
+        return AuthCredentials(["authenticated"]), Peer(
             identity=payload["peer"],
             token=token,
             payload=payload,
-            origin=conn.client.host,
+            origin=host,
         )
 
     async def authenticate(
@@ -93,4 +66,4 @@ async def authenticate(
             binascii.Error,
         ) as exc:
             self.logger.error(f"Invalid token for peer ({exc})")
-            raise AuthenticationError(f"access not allowed for {conn.client.host}")
+            raise AuthenticationError(f"access not allowed for {conn.client.host}")  # type: ignore

horao/auth/permissions.py

@@ -0,0 +1,86 @@
+# -*- coding: utf-8 -*-#
+"""Permission classes and functions for the role-based access control module."""
+from __future__ import annotations
+
+from abc import ABC
+from enum import Enum, auto
+from typing import Dict, TypeVar
+
+RT = TypeVar("RT")
+
+
+class Namespace(Enum):
+    """Namespaces define where a permission applies"""
+
+    System = auto()
+    User = auto()
+    Peer = auto()
+
+
+class Permission(Enum):
+    """We have two permissions: Read and Write, Write implies read."""
+
+    Read = auto()
+    Write = auto()
+
+
+class Permissions(ABC):
+    def __init__(self, name: str, permissions: Dict[Namespace, Permission]):
+        self.name = name
+        self.permissions: Dict[Namespace, Permission] = permissions
+
+    def __len__(self):
+        return len(self.permissions.keys())
+
+    def __repr__(self):
+        return f"<Permissions {self.name}>"
+
+    def __iter__(self):
+        for permission in self.permissions:
+            yield permission
+
+    def can_read(self, namespace: Namespace):
+        if namespace not in self.permissions.keys():
+            return False
+        return (
+            self.permissions[namespace] == Permission.Read
+            or self.permissions[namespace] == Permission.Write
+        )
+
+    def can_write(self, namespace: Namespace):
+        if namespace not in self.permissions.keys():
+            return False
+        return self.permissions[namespace] == Permission.Write
+
+
+class AdministratorPermissions(Permissions):
+    def __init__(self):
+        super().__init__(
+            "System Administrator",
+            {Namespace.System: Permission.Write, Namespace.User: Permission.Read},
+        )
+
+    def __str__(self):
+        return self.name
+
+
+class PeerPermissions(Permissions):
+    def __init__(self):
+        super().__init__(
+            "Peer Node",
+            {Namespace.Peer: Permission.Write},
+        )
+
+    def __str__(self):
+        return self.name
+
+
+class TenantPermissions(Permissions):
+    def __init__(self):
+        super().__init__(
+            "TenantOwner",
+            {Namespace.System: Permission.Read, Namespace.User: Permission.Write},
+        )
+
+    def __str__(self):
+        return self.name

horao/auth/validate.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from functools import wraps
+from typing import Callable
+
+from starlette.requests import Request
+
+from horao.auth.error import UnauthorizedError
+from horao.auth.permissions import RT, Namespace, Permission
+from horao.auth.roles import User
+
+
+def permission_required(
+    namespace: Namespace, permission: Permission
+) -> Callable[[Callable[..., RT]], Callable[..., RT]]:
+    """
+    Decorator to check if a user has the permission to access a resource.
+    :param namespace: namespace to check
+    :param permission: permission to check
+    :return: function call if the user has the permission
+    :raises: UnauthorizedError if the user does not have the permission
+    """
+
+    def decorator(func: Callable[..., RT]) -> Callable[..., RT]:
+        @wraps(func)
+        async def wrapper(*args: str, **kwargs: int) -> RT:
+            for arg in args:
+                if isinstance(arg, Request):
+                    if not arg.user:
+                        raise UnauthorizedError(func, *args, **kwargs)
+                    if isinstance(arg.user, User):
+                        if permission.Write and any(
+                            [p for p in arg.user.permissions if p.can_write(namespace)]
+                        ):
+                            return await func(*args, **kwargs)
+                        if permission.Read and any(
+                            [p for p in arg.user.permissions if p.can_read(namespace)]
+                        ):
+                            return await func(*args, **kwargs)
+            raise UnauthorizedError(func, *args, **kwargs)
+
+        return wrapper  # type: ignore
+
+    return decorator

horao/conceptual/claim.py

@@ -11,13 +11,6 @@
 from horao.physical.computer import Computer
 from horao.physical.hardware import Hardware
 from horao.physical.storage import StorageType
-from horao.rbac.roles import (
-    Delegate,
-    NetworkEngineer,
-    SecurityEngineer,
-    SystemEngineer,
-    TenantOwner,
-)
 
 
 class Claim(ABC):
@@ -84,7 +77,6 @@ def __init__(
         start: datetime,
         end: datetime,
         reason: str,
-        operator: SecurityEngineer | SystemEngineer | NetworkEngineer,
         target: List[DataCenter | DataCenterNetwork | Computer | Hardware],
     ):
         """
@@ -98,7 +90,6 @@ def __init__(
         """
         super().__init__(name, start, end)
         self.reason = reason
-        self.operator = operator
         self.target = target
 
     def __eq__(self, other) -> bool:
@@ -107,14 +98,11 @@ def __eq__(self, other) -> bool:
         return (
             super().__eq__(other)
             and self.reason == other.reason
-            and self.operator == other.operator
             and self.target == other.target
         )
 
     def __hash__(self):
-        return hash(
-            (self.name, self.start, self.end, self.reason, self.operator, self.target)
-        )
+        return hash((self.name, self.start, self.end, self.reason, self.target))
 
 
 class Reservation(Claim):