schema.py

"""Schemas for main router.

Copyright (c) 2024 MultiFactor
License: https://github.com/MultiDirectoryLab/MultiDirectory/blob/main/LICENSE
"""

from typing import final

from dishka import AsyncContainer
from pydantic import BaseModel, Field, PrivateAttr, SecretStr
from sqlalchemy.sql.elements import ColumnElement, UnaryExpression

from entities import Directory
from ldap_protocol.filter_interpreter import (
    Filter,
    FilterInterpreterProtocol,
    StringFilterInterpreter,
)
from ldap_protocol.ldap_requests import SearchRequest as LDAPSearchRequest
from ldap_protocol.ldap_responses import SearchResultDone, SearchResultEntry
from ldap_protocol.utils.const import GRANT_DN_STRING


class SearchRequest(LDAPSearchRequest):
    """Search request for web api."""

    filter: str = Field(..., examples=["(objectClass=*)"])  # type: ignore

    _filter_interpreter: FilterInterpreterProtocol = PrivateAttr(
        default_factory=StringFilterInterpreter,
    )

    def _cast_filter(self) -> UnaryExpression | ColumnElement:
        """Cast str filter to sa sql."""
        filter_ = self.filter.lower().replace("objectcategory", "objectclass")
        return self._filter_interpreter.cast_to_sql(
            Filter.parse(filter_).simplify(),
        )

    @staticmethod
    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
        return str(directory.object_guid)

    @final
    async def handle_api(  # type: ignore
        self,
        container: AsyncContainer,
    ) -> list[SearchResultEntry | SearchResultDone]:
        """Get all responses."""
        return await self._handle_api(container)  # type: ignore


class SearchResponse(SearchResultDone):
    """Search response for web api."""

    search_result: list[SearchResultEntry]


class KerberosSetupRequest(BaseModel):
    """Kerberos setup data."""

    krbadmin_password: SecretStr
    admin_password: SecretStr
    stash_password: SecretStr


class PrincipalAddRequest(BaseModel):
    """Request schema for POST /principal/add."""

    principal_name: str
    algorithms: list[str] | None = None
    password: str | None = None


class KtaddRequest(BaseModel):
    """Request schema for POST /ktadd."""

    names: list[str]
    is_rand_key: bool = False


class ModifyPrincipalRequest(BaseModel):
    """Request schema for PUT /principal (full modify)."""

    principal_name: str
    new_name: str | None = None
    algorithms: list[str] | None = None
    password: str | None = None


class PrimaryGroupRequest(BaseModel):
    """Request schema for setting primary group."""

    directory_dn: GRANT_DN_STRING
    group_dn: GRANT_DN_STRING


class PrimaryGroupPathDNResponse(BaseModel):
    """Response schema for getting group path DN by primary group ID."""

    path_dn: str