Update objectsid logic 1051

app/api/main/schema.py

@@ -38,8 +38,11 @@ def _cast_filter(self) -> UnaryExpression | ColumnElement:
         )
 
     @staticmethod
-    def get_directory_sid(directory: Directory) -> str:  # type: ignore
-        return directory.object_sid
+    def get_directory_sid(directory: Directory) -> str | None:  # type: ignore
+        for attr in getattr(directory, "attributes", []):
+            if attr.name and attr.name.lower() == "objectsid" and attr.value:
+                return attr.value
+        return None
 
     @staticmethod
     def get_directory_guid(directory: Directory) -> str:  # type: ignore

app/constants.py

@@ -6,11 +6,12 @@
 
 from typing import TypedDict
 
-from enums import EntityTypeNames, SamAccountTypeCodes
+from enums import EntityTypeNames, SamAccountTypeCodes, SecurityPrincipalRid
 
 GROUPS_CONTAINER_NAME = "Groups"
 COMPUTERS_CONTAINER_NAME = "Computers"
 USERS_CONTAINER_NAME = "Users"
+SYSTEM_CONTAINER_NAME = "System"
 DOMAIN_CONTROLLERS_OU_NAME = "Domain Controllers"
 
 READ_ONLY_GROUP_NAME = "read-only"
@@ -293,6 +294,14 @@ class EntityTypeData(TypedDict):
 
 
 FIRST_SETUP_DATA = [
+    {
+        "name": SYSTEM_CONTAINER_NAME,
+        "object_class": "organizationalUnit",
+        "attributes": {
+            "objectClass": ["top", "container"],
+        },
+        "children": [],
+    },
     {
         "name": GROUPS_CONTAINER_NAME,
         "object_class": "container",
@@ -314,7 +323,7 @@ class EntityTypeData(TypedDict):
                     ],
                     "gidNumber": ["512"],
                 },
-                "objectSid": 512,
+                "objectSid": SecurityPrincipalRid.DOMAIN_ADMINS,
             },
             {
                 "name": DOMAIN_USERS_GROUP_NAME,
@@ -329,7 +338,7 @@ class EntityTypeData(TypedDict):
                     ],
                     "gidNumber": ["513"],
                 },
-                "objectSid": 513,
+                "objectSid": SecurityPrincipalRid.DOMAIN_USERS,
             },
             {
                 "name": READ_ONLY_GROUP_NAME,
@@ -344,7 +353,7 @@ class EntityTypeData(TypedDict):
                     ],
                     "gidNumber": ["521"],
                 },
-                "objectSid": 521,
+                "objectSid": SecurityPrincipalRid.DOMAIN_READ_ONLY,
             },
             {
                 "name": DOMAIN_COMPUTERS_GROUP_NAME,
@@ -359,7 +368,7 @@ class EntityTypeData(TypedDict):
                     ],
                     "gidNumber": ["515"],
                 },
-                "objectSid": 515,
+                "objectSid": SecurityPrincipalRid.DOMAIN_COMPUTERS,
             },
         ],
     },

app/entities.py

@@ -231,7 +231,6 @@ class Directory:
     search_fields: ClassVar[dict[str, str]] = {
         "name": "name",
         "objectguid": "objectGUID",
-        "objectsid": "objectSid",
     }
     ro_fields: ClassVar[set[str]] = {
         "uid",
@@ -277,12 +276,17 @@ def create_path(
 
     @property
     def relative_id(self) -> str:
-        """Get RID from objectSid.
+        """Get RID from objectSid attribute.
 
         Relative Identifier (RID) is the last sub-authority value of a SID.
         """
-        if "-" in self.object_sid:
-            return self.object_sid.split("-")[-1]
+        attrs = self.__dict__.get("attributes")
+        if not attrs:
+            return ""
+
+        for attr in attrs:
+            if attr.name and attr.name.lower() == "objectsid" and attr.value:
+                return attr.value.split("-")[-1]
         return ""
 
     @property

app/enums.py

@@ -69,6 +69,8 @@ class EntityTypeNames(StrEnum):
     KRB_CONTAINER = "KRB Container"
     KRB_PRINCIPAL = "KRB Principal"
     KRB_REALM_CONTAINER = "KRB Realm Container"
+    RID_MANAGER = "RID Manager"
+    RID_SET = "RID Set"
 
 
 class KindType(StrEnum):
@@ -280,3 +282,22 @@ class SamAccountTypeCodes(IntEnum):
     def to_hex(value: int) -> str:
         """Convert decimal value to hex string."""
         return hex(value)
+
+
+class SidPrefix(StrEnum):
+    """SID prefix."""
+
+    DOMAIN_IDENTIFIER = "S-1-5-21"
+    BUILT_IN_DOMAIN = "S-1-5-32"
+
+
+class SecurityPrincipalRid(IntEnum):
+    ADMINISTRATOR = 500
+    GUESTS = 501
+    KRBTGT = 502
+    DOMAIN_ADMINS = 512
+    DOMAIN_USERS = 513
+    DOMAIN_GUESTS = 514
+    DOMAIN_COMPUTERS = 515
+    DOMAIN_CONTROLLERS = 516
+    DOMAIN_READ_ONLY = 521

app/extra/scripts/add_domain_controller.py

@@ -11,12 +11,11 @@
 from config import Settings
 from constants import DOMAIN_CONTROLLERS_OU_NAME
 from entities import Attribute, Directory
-from enums import SamAccountTypeCodes
+from enums import SamAccountTypeCodes, SecurityPrincipalRid
 from ldap_protocol.ldap_schema.entity_type_dao import EntityTypeDAO
 from ldap_protocol.objects import UserAccountControlFlag
+from ldap_protocol.rid_manager import ObjectSIDUseCase
 from ldap_protocol.roles.role_use_case import RoleUseCase
-from ldap_protocol.utils.helpers import create_object_sid
-from ldap_protocol.utils.queries import get_base_directories
 from repo.pg.tables import queryable_attr as qa
 
 
@@ -25,8 +24,8 @@ async def _add_domain_controller(
     role_use_case: RoleUseCase,
     entity_type_dao: EntityTypeDAO,
     settings: Settings,
-    domain: Directory,
     dc_ou_dir: Directory,
+    object_sid_use_case: ObjectSIDUseCase,
 ) -> None:
     dc_directory = Directory(
         object_class="",
@@ -38,7 +37,10 @@ async def _add_domain_controller(
     await session.flush()
 
     dc_directory.parent_id = dc_ou_dir.id
-    dc_directory.object_sid = create_object_sid(domain, dc_directory.id)
+    await object_sid_use_case.add(
+        directory=dc_directory,
+        rid=SecurityPrincipalRid.DOMAIN_CONTROLLERS,
+    )
     await session.flush()
 
     attributes = [
@@ -101,14 +103,10 @@ async def add_domain_controller(
     settings: Settings,
     role_use_case: RoleUseCase,
     entity_type_dao: EntityTypeDAO,
+    object_sid_use_case: ObjectSIDUseCase,
 ) -> None:
     logger.info("Adding domain controller.")
 
-    domains = await get_base_directories(session)
-    if not domains:
-        logger.debug("Cannot get base directory")
-        return
-
     domain_controllers_ou = await session.scalar(
         select(Directory).where(
             qa(Directory.name) == DOMAIN_CONTROLLERS_OU_NAME,
@@ -138,8 +136,8 @@ async def add_domain_controller(
         role_use_case=role_use_case,
         entity_type_dao=entity_type_dao,
         settings=settings,
-        domain=domains[0],
         dc_ou_dir=domain_controllers_ou,
+        object_sid_use_case=object_sid_use_case,
     )
 
     logger.debug("Domain controller added.")

app/ioc.py

@@ -153,6 +153,16 @@
     PasswordBanWordUseCases,
     UserPasswordHistoryUseCases,
 )
+from ldap_protocol.rid_manager import (
+    ObjectSIDGateway,
+    ObjectSIDUseCase,
+    RIDManagerGateway,
+    RIDManagerSetupGateway,
+    RIDManagerSetupUseCase,
+    RIDManagerUseCase,
+    RIDSetGateway,
+    RIDSetUseCase,
+)
 from ldap_protocol.roles.access_manager import AccessManager
 from ldap_protocol.roles.ace_dao import AccessControlEntryDAO
 from ldap_protocol.roles.role_dao import RoleDAO
@@ -564,6 +574,21 @@ def get_dhcp_mngr(
     rootdse_reader = provide(RootDSEReader, scope=Scope.REQUEST)
     dcinfo_reader = provide(DCInfoReader, scope=Scope.REQUEST)
 
+    rid_manager_gateway = provide(RIDManagerGateway, scope=Scope.REQUEST)
+    rid_manager_setup_gateway = provide(
+        RIDManagerSetupGateway,
+        scope=Scope.REQUEST,
+    )
+    rid_manager_use_case = provide(RIDManagerUseCase, scope=Scope.REQUEST)
+    rid_manager_setup_use_case = provide(
+        RIDManagerSetupUseCase,
+        scope=Scope.REQUEST,
+    )
+    object_sid_gateway = provide(ObjectSIDGateway, scope=Scope.REQUEST)
+    object_sid_use_case = provide(ObjectSIDUseCase, scope=Scope.REQUEST)
+    rid_set_gateway = provide(RIDSetGateway, scope=Scope.REQUEST)
+    rid_set_use_case = provide(RIDSetUseCase, scope=Scope.REQUEST)
+
 
 class LDAPContextProvider(Provider):
     """Context provider."""

app/ldap_protocol/auth/setup_gateway.py

@@ -12,12 +12,13 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from entities import Attribute, Directory, Group, NetworkPolicy, User
+from enums import SidPrefix
 from ldap_protocol.ldap_schema.attribute_value_validator import (
     AttributeValueValidator,
 )
 from ldap_protocol.ldap_schema.entity_type_dao import EntityTypeDAO
+from ldap_protocol.rid_manager import ObjectSIDUseCase
 from ldap_protocol.utils.async_cache import base_directories_cache
-from ldap_protocol.utils.helpers import create_object_sid, generate_domain_sid
 from ldap_protocol.utils.queries import get_domain_object_class
 from password_utils import PasswordUtils
 from repo.pg.tables import queryable_attr as qa
@@ -32,6 +33,7 @@ def __init__(
         password_utils: PasswordUtils,
         entity_type_dao: EntityTypeDAO,
         attribute_value_validator: AttributeValueValidator,
+        object_sid_use_case: ObjectSIDUseCase,
     ) -> None:
         """Initialize Setup use case.
 
@@ -43,6 +45,7 @@ def __init__(
         self._password_utils = password_utils
         self._entity_type_dao = entity_type_dao
         self._attribute_value_validator = attribute_value_validator
+        self._object_sid_use_case = object_sid_use_case
 
     async def is_setup(self) -> bool:
         """Check if setup is performed.
@@ -61,21 +64,9 @@ async def setup_enviroment(
         *,
         data: list,
         is_system: bool = True,
-        dn: str = "multifactor.dev",
+        domain: Directory,
     ) -> None:
         """Create directories and users for enviroment."""
-        cat_result = await self._session.execute(select(Directory))
-        if cat_result.scalar_one_or_none():
-            logger.warning("dev data already set up")
-            return
-
-        domain = Directory(name=dn, object_class="domain")
-        domain.is_system = True
-        domain.object_sid = generate_domain_sid()
-        domain.path = [f"dc={path}" for path in reversed(dn.split("."))]
-        domain.depth = len(domain.path)
-        domain.rdname = ""
-
         async with self._session.begin_nested():
             self._session.add(domain)
             self._session.add(
@@ -122,6 +113,28 @@ async def setup_enviroment(
             logger.error(traceback.format_exc())
             raise
 
+    async def is_base_domain_created(self) -> bool:
+        """Check if base domain is created."""
+        cat_result = await self._session.execute(select(Directory))
+        if cat_result.scalar_one_or_none():
+            logger.warning("dev data already set up")
+            return True
+        return False
+
+    async def create_base_domain(
+        self,
+        dn: str = "multifactor.dev",
+    ) -> Directory:
+        """Create base domain."""
+        domain = Directory(name=dn, object_class="domain")
+        domain.is_system = True
+        domain.path = [f"dc={path}" for path in reversed(dn.split("."))]
+        domain.depth = len(domain.path)
+        domain.rdname = ""
+        self._session.add(domain)
+        await self._session.flush()
+        return domain
+
     async def create_dir(
         self,
         data: dict,
@@ -151,11 +164,12 @@ async def create_dir(
             ),
         )
 
-        dir_.object_sid = create_object_sid(
-            domain,
-            rid=data.get("objectSid", dir_.id),
-            reserved="objectSid" in data,
-        )
+        if "objectSid" in data:
+            await self._object_sid_use_case.add(
+                directory=dir_,
+                rid=int(data["objectSid"]),
+                sid_prefix=SidPrefix.BUILT_IN_DOMAIN,
+            )
 
         if dir_.object_class == "group":
             group = Group(directory_id=dir_.id)