Event log refactor 290

app/alembic/script.py.mako

@@ -17,8 +17,10 @@ depends_on = ${repr(depends_on)}
 
 
 def upgrade() -> None:
+    """Upgrade."""
     ${upgrades if upgrades else "pass"}
 
 
 def downgrade() -> None:
+    """Downgrade."""
     ${downgrades if downgrades else "pass"}

app/api/main/adapters/kerberos.py

@@ -22,7 +22,7 @@
     KerberosUnavailableError,
 )
 from ldap_protocol.kerberos.service import KerberosService
-from ldap_protocol.ldap_schema.entity_type_dao import EntityTypeDAO
+from ldap_protocol.ldap_requests.contexts import LDAPAddRequestContext
 
 P = ParamSpec("P")
 R = TypeVar("R")
@@ -71,7 +71,7 @@ async def setup_krb_catalogue(
         mail: str,
         krbadmin_password: SecretStr,
         ldap_session: LDAPSession,
-        entity_type_dao: EntityTypeDAO,
+        ctx: LDAPAddRequestContext,
     ) -> None:
         """Create Kerberos structure in the LDAP directory.
 
@@ -83,7 +83,7 @@ async def setup_krb_catalogue(
             mail,
             krbadmin_password,
             ldap_session,
-            entity_type_dao,
+            ctx,
         )
 
     async def setup_kdc(

app/api/main/krb5_router.py

@@ -20,7 +20,7 @@
 from api.main.schema import KerberosSetupRequest
 from ldap_protocol.dialogue import LDAPSession, UserSchema
 from ldap_protocol.kerberos import KerberosState
-from ldap_protocol.ldap_schema.entity_type_dao import EntityTypeDAO
+from ldap_protocol.ldap_requests.contexts import LDAPAddRequestContext
 from ldap_protocol.utils.const import EmailStr
 
 from .utils import get_ldap_session
@@ -42,7 +42,7 @@ async def setup_krb_catalogue(
     mail: Annotated[EmailStr, Body()],
     krbadmin_password: Annotated[SecretStr, Body()],
     ldap_session: Annotated[LDAPSession, Depends(get_ldap_session)],
-    entity_type_dao: FromDishka[EntityTypeDAO],
+    ctx: FromDishka[LDAPAddRequestContext],
     kerberos_adapter: FromDishka[KerberosFastAPIAdapter],
 ) -> None:
     """Generate tree for kdc/kadmin.
@@ -56,7 +56,7 @@ async def setup_krb_catalogue(
         mail,
         krbadmin_password,
         ldap_session,
-        entity_type_dao,
+        ctx,
     )
 
 

app/ioc.py

@@ -31,6 +31,16 @@
 from ldap_protocol.kerberos.ldap_structure import KRBLDAPStructureManager
 from ldap_protocol.kerberos.service import KerberosService
 from ldap_protocol.kerberos.template_render import KRBTemplateRenderer
+from ldap_protocol.ldap_requests.contexts import (
+    LDAPAddRequestContext,
+    LDAPBindRequestContext,
+    LDAPDeleteRequestContext,
+    LDAPExtendedRequestContext,
+    LDAPModifyDNRequestContext,
+    LDAPModifyRequestContext,
+    LDAPSearchRequestContext,
+    LDAPUnbindRequestContext,
+)
 from ldap_protocol.ldap_schema.attribute_type_dao import AttributeTypeDAO
 from ldap_protocol.ldap_schema.entity_type_dao import EntityTypeDAO
 from ldap_protocol.ldap_schema.object_class_dao import ObjectClassDAO
@@ -206,7 +216,44 @@ async def get_session_storage(
     role_use_case = provide(RoleUseCase, scope=Scope.REQUEST)
 
 
-class HTTPProvider(Provider):
+class LDAPContextProvider(Provider):
+    """Context provider."""
+
+    add_request_context = provide(
+        LDAPAddRequestContext,
+        scope=Scope.REQUEST,
+    )
+    bind_request_context = provide(
+        LDAPBindRequestContext,
+        scope=Scope.REQUEST,
+    )
+    delete_request_context = provide(
+        LDAPDeleteRequestContext,
+        scope=Scope.REQUEST,
+    )
+    extended_request_context = provide(
+        LDAPExtendedRequestContext,
+        scope=Scope.REQUEST,
+    )
+    modify_request_context = provide(
+        LDAPModifyRequestContext,
+        scope=Scope.REQUEST,
+    )
+    modify_dn_request_context = provide(
+        LDAPModifyDNRequestContext,
+        scope=Scope.REQUEST,
+    )
+    search_request_context = provide(
+        LDAPSearchRequestContext,
+        scope=Scope.REQUEST,
+    )
+    unbind_request_context = provide(
+        LDAPUnbindRequestContext,
+        scope=Scope.REQUEST,
+    )
+
+
+class HTTPProvider(LDAPContextProvider):
     """HTTP LDAP session."""
 
     scope = Scope.REQUEST
@@ -244,7 +291,7 @@ def get_krb_template_render(
     krb_ldap_manager = provide(KRBLDAPStructureManager, scope=Scope.REQUEST)
 
 
-class LDAPServerProvider(Provider):
+class LDAPServerProvider(LDAPContextProvider):
     """Provider with session scope."""
 
     scope = Scope.SESSION

app/ldap_protocol/asn1parser.py

@@ -6,11 +6,10 @@
 
 from contextlib import suppress
 from dataclasses import dataclass
-from enum import IntEnum
-from typing import Annotated, Generic, TypeVar
+from enum import IntEnum, StrEnum
+from typing import Generic, TypeVar
 
 from asn1 import Classes, Decoder, Encoder, Numbers, Tag, Types
-from pydantic import AfterValidator
 
 
 class TagNumbers(IntEnum):
@@ -276,11 +275,15 @@ def asn1todict(decoder: Decoder) -> list[ASN1Row]:
     return out
 
 
-def _validate_oid(oid: str) -> str:
-    """Validate ldap oid with regex."""
-    if not Encoder._re_oid.match(oid):  # noqa SLF001
-        raise ValueError("Invalid LDAPOID")
-    return oid
+class LDAPOID(StrEnum):
+    """Enum for LDAP OIDs."""
 
+    PASSWORD_MODIFY = "1.3.6.1.4.1.4203.1.11.1"  # noqa
+    WHOAMI = "1.3.6.1.4.1.4203.1.11.3"
+    START_TLS = "1.3.6.1.4.1.1466.20037"
+    PAGED_RESULTS = "1.2.840.113556.1.4.319"
 
-LDAPOID = Annotated[str, AfterValidator(_validate_oid)]
+    @classmethod
+    def has_value(cls, value: str) -> bool:
+        """Check if value is a valid LDAPOID."""
+        return Encoder._re_oid.match(value) is not None  # noqa: SLF001

app/ldap_protocol/dependency.py

@@ -4,26 +4,26 @@
 License: https://github.com/MultiDirectoryLab/MultiDirectory/blob/main/LICENSE
 """
 
-from functools import partial, wraps
-from typing import Callable, TypeVar, get_type_hints
+import contextlib
+from typing import Any, Callable, TypeVar, get_type_hints
 
 from dishka import AsyncContainer
+from dishka.exceptions import NoFactoryError
 
 T = TypeVar("T", bound=Callable)
 
 
-async def resolve_deps(func: T, container: AsyncContainer) -> T:
+async def resolve_deps(func: T, container: AsyncContainer) -> dict[str, Any]:
     """Provide async dependencies.
 
     :param T func: Awaitable
     :param AsyncContainer container: IoC container
-    :return T: Awaitable
+    :return dict[str, Any]: kwargs for func
     """
-    hints = get_type_hints(func)
-    del hints["return"]
     kwargs = {}
 
-    for arg_name, hint in hints.items():
-        kwargs[arg_name] = await container.get(hint)
+    for arg_name, hint in get_type_hints(func).items():
+        with contextlib.suppress(NoFactoryError):
+            kwargs[arg_name] = await container.get(hint)
 
-    return wraps(func)(partial(func, **kwargs))  # type: ignore
+    return kwargs

app/ldap_protocol/kerberos/ldap_structure.py

@@ -7,17 +7,14 @@
 from sqlalchemy import delete, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from ldap_protocol.dialogue import LDAPSession
 from ldap_protocol.kerberos.exceptions import KerberosConflictError
 from ldap_protocol.ldap_requests import AddRequest
-from ldap_protocol.ldap_schema.entity_type_dao import EntityTypeDAO
+from ldap_protocol.ldap_requests.contexts import LDAPAddRequestContext
 from ldap_protocol.roles.access_manager import AccessManager
 from ldap_protocol.roles.role_use_case import RoleUseCase
 from ldap_protocol.utils.queries import get_filter_from_path
 from models import Directory
 
-from .base import AbstractKadmin
-
 
 class KRBLDAPStructureManager:
     """Manager for Kerberos-related LDAP structure operations."""
@@ -43,9 +40,7 @@ async def create_kerberos_structure(
         group: AddRequest,
         services: AddRequest,
         krb_user: AddRequest,
-        ldap_session: LDAPSession,
-        kadmin: AbstractKadmin,
-        entity_type_dao: EntityTypeDAO,
+        ctx: LDAPAddRequestContext,
     ) -> None:
         """Create Kerberos structure in the LDAP directory.
 
@@ -62,36 +57,9 @@ async def create_kerberos_structure(
         """
         async with self._session.begin_nested():
             results = (
-                await anext(
-                    services.handle(
-                        self._session,
-                        ldap_session,
-                        kadmin,
-                        entity_type_dao,
-                        self._access_manager,
-                        self._role_use_case,
-                    )
-                ),
-                await anext(
-                    group.handle(
-                        self._session,
-                        ldap_session,
-                        kadmin,
-                        entity_type_dao,
-                        self._access_manager,
-                        self._role_use_case,
-                    )
-                ),
-                await anext(
-                    krb_user.handle(
-                        self._session,
-                        ldap_session,
-                        kadmin,
-                        entity_type_dao,
-                        self._access_manager,
-                        self._role_use_case,
-                    )
-                ),
+                await anext(services.handle(ctx)),
+                await anext(group.handle(ctx)),
+                await anext(krb_user.handle(ctx)),
             )
             await self._session.flush()
 

app/ldap_protocol/kerberos/service.py

@@ -22,7 +22,7 @@
     KerberosUnavailableError,
 )
 from ldap_protocol.ldap_requests import AddRequest
-from ldap_protocol.ldap_schema.entity_type_dao import EntityTypeDAO
+from ldap_protocol.ldap_requests.contexts import LDAPAddRequestContext
 from ldap_protocol.utils.queries import get_base_directories, get_dn_by_id
 
 from .base import AbstractKadmin, KerberosState, KRBAPIError
@@ -66,21 +66,21 @@ async def setup_krb_catalogue(
         mail: str,
         krbadmin_password: SecretStr,
         ldap_session: LDAPSession,
-        entity_type_dao: EntityTypeDAO,
+        ctx: LDAPAddRequestContext,
     ) -> None:
         """Create Kerberos structure in the LDAP directory.
 
         Args:
             mail (str): Email for krbadmin.
             krbadmin_password (SecretStr): Password for krbadmin.
-            ldap_session (LDAPSession): LDAP session.
-            entity_type_dao (EntityTypeDAO): DAO for entity types.
-            access_manager (AccessManager): Access control manager.
+            ldap_session (LDAPSession): ldap session.
+            ctx (LDAPAddRequestContext): context for add request.
 
         Raises:
             KerberosConflictError: On structure creation conflict.
 
         """
+        ctx.ldap_session = ldap_session
         base_dn, _ = await self._get_base_dn()
         dns = self._build_kerberos_admin_dns(base_dn)
         add_requests = self._build_add_requests(
@@ -92,9 +92,7 @@ async def setup_krb_catalogue(
             add_requests.group,
             add_requests.services,
             add_requests.krb_user,
-            ldap_session,
-            self._kadmin,
-            entity_type_dao,
+            ctx,
         )
 
     async def _get_base_dn(self) -> tuple[str, str]:

app/ldap_protocol/ldap_requests/abandon.py

@@ -8,14 +8,15 @@
 from typing import AsyncGenerator, ClassVar
 
 from ldap_protocol.asn1parser import ASN1Row
+from ldap_protocol.objects import ProtocolRequests
 
 from .base import BaseRequest
 
 
 class AbandonRequest(BaseRequest):
     """Abandon protocol."""
 
-    PROTOCOL_OP: ClassVar[int] = 16
+    PROTOCOL_OP: ClassVar[int] = ProtocolRequests.ABANDON
     message_id: int
 
     @classmethod