Conditional Access policies must not reference deleted users, groups, or roles

from prowler.lib.check.models import Check, CheckReportM365
from prowler.providers.m365.services.entra.entra_client import entra_client
from prowler.providers.m365.services.entra.entra_service import (
    CONDITIONAL_ACCESS_SENTINEL_IDS,
    ConditionalAccessPolicyState,
)


class entra_conditional_access_policy_no_deleted_object_references(Check):
    """
    Ensure Conditional Access policies do not reference deleted directory objects.

    Stale references to deleted users, groups, or directory roles silently change
    the runtime behavior of a Conditional Access policy: include* references
    shrink enforcement scope, exclude* references can change exemption logic.
    Either way, the policy stops behaving the way the operator believes it does.

    The directory-object existence check runs once at service init time and is
    cached on the entra client. This check reads from that cache and reports any
    policy whose users/groups/roles inclusion or exclusion collections name an
    identifier that no longer resolves in Microsoft Entra ID.

    Identifiers whose Graph lookup failed with a non-404 error (5xx, throttling,
    insufficient permissions) are cached separately: they could not be verified
    as present or deleted, so the policy is reported as MANUAL rather than being
    silently treated as clean.

    - PASS: The policy references no deleted users, groups, or roles.
    - FAIL: The policy references at least one deleted user, group, or role.
    - MANUAL: At least one referenced identifier could not be resolved due to a
      transient Graph error, so the policy could not be fully evaluated.
    """

    def execute(self) -> list[CheckReportM365]:
        findings = []
        unresolved = entra_client.unresolved_directory_object_references
        errored = entra_client.errored_directory_object_references

        for policy_id, policy in entra_client.conditional_access_policies.items():
            report = CheckReportM365(
                metadata=self.metadata(),
                resource=policy,
                resource_name=policy.display_name,
                resource_id=policy_id,
            )

            orphans = self._collect_references_in(policy, unresolved)
            unverified = self._collect_references_in(policy, errored)

            if orphans:
                # A confirmed deletion takes precedence over unverified ones.
                report.status = "FAIL"
                report.status_extended = self._format_failure(
                    policy.display_name, orphans, unverified, policy.state
                )
            elif unverified:
                # Nothing confirmed deleted, but we could not verify every
                # reference — do not claim the policy is clean.
                report.status = "MANUAL"
                report.status_extended = self._format_manual(
                    policy.display_name, unverified
                )
            else:
                report.status = "PASS"
                report.status_extended = (
                    f"Conditional Access policy {policy.display_name} references no "
                    f"deleted directory objects."
                )

            findings.append(report)

        return findings

    @staticmethod
    def _collect_references_in(policy, id_set):
        """Walk the six identifier collections and return references in ``id_set``.

        Args:
            policy: The Conditional Access policy to inspect.
            id_set: Set of ``(type, id)`` pairs to match references against.

        Returns:
            list[tuple[str, str, str]]: ``(type, id, side)`` tuples where
                ``type`` is one of ``user|group|role``, ``id`` is the Graph
                identifier, and ``side`` is one of ``include|exclude``.
        """
        if not policy.conditions or not policy.conditions.user_conditions:
            return []

        uc = policy.conditions.user_conditions
        collections = (
            ("user", "include", uc.included_users),
            ("user", "exclude", uc.excluded_users),
            ("group", "include", uc.included_groups),
            ("group", "exclude", uc.excluded_groups),
            ("role", "include", uc.included_roles),
            ("role", "exclude", uc.excluded_roles),
        )

        matches = []
        for type_, side, identifiers in collections:
            for identifier in identifiers:
                if identifier in CONDITIONAL_ACCESS_SENTINEL_IDS:
                    continue
                if (type_, identifier) in id_set:
                    matches.append((type_, identifier, side))
        return matches

    @staticmethod
    def _group_by_type(references):
        """Group ``(type, id, side)`` references into a deterministic message part."""
        by_type = {"user": [], "group": [], "role": []}
        for type_, identifier, side in references:
            by_type[type_].append(f"{identifier} ({side})")

        parts = []
        for type_ in ("user", "group", "role"):
            if by_type[type_]:
                joined = ", ".join(sorted(by_type[type_]))
                parts.append(f"{type_}s: {joined}")
        return "; ".join(parts)

    @classmethod
    def _format_failure(cls, display_name, orphans, unverified, state=None):
        # Surface report-only mode explicitly: the stale references are not yet
        # enforced, but become live the moment the policy is turned on.
        report_only = (
            " The policy is in report-only mode, so these references are not "
            "enforced yet but will take effect once it is enabled."
            if state == ConditionalAccessPolicyState.ENABLED_FOR_REPORTING
            else ""
        )

        # If some references also could not be resolved, say so rather than
        # implying the rest of the policy is fully clean.
        unverified_note = (
            f" Additionally, {len(unverified)} reference(s) could not be verified "
            f"due to transient Microsoft Graph errors."
            if unverified
            else ""
        )

        return (
            f"Conditional Access policy {display_name} references "
            f"{len(orphans)} deleted directory object(s) — "
            f"{cls._group_by_type(orphans)}.{report_only}{unverified_note}"
        )

    @classmethod
    def _format_manual(cls, display_name, unverified):
        return (
            f"Conditional Access policy {display_name} could not be fully evaluated: "
            f"{len(unverified)} reference(s) could not be resolved due to transient "
            f"Microsoft Graph errors (5xx, throttling, or insufficient permissions) — "
            f"{cls._group_by_type(unverified)}. Re-run the scan or review the policy "
            f"manually."
        )