"""Auditor-side adapters that READ upstream state from configured
integrations.

The existing ``app/integrations/drivers/*.py`` modules are
provisioning-shaped — they DRIVE downstream actions (grant seat,
revoke account, force-logout). The auditor needs the orthogonal
direction: enumerate state the auditor cares about (MFA enrollment,
admin role assignments, signin events, public ingress rules, certs,
PATs).

Rather than bloat each provisioning driver, those auditor reads live
here. Each adapter reads its credentials from the existing
``integrations`` row resolution helper (``integrations.service``),
constructs a fresh API client, and returns a normalised list of dicts
the caller (auditor service) writes into the ``auditor_*`` tables.

Failure semantics:

  * ``IntegrationMissingError`` raised if the kind/provider has no
    configured integration. The caller logs and returns an empty list
    — that is the *real* state of the world, not a fallback.

  * Underlying API exceptions propagate. The caller wraps them in
    audit log entries so the auditor sees "google_signin_pull failed:
    HTTP 403, missing scope X" instead of silently zero new events.

This module talks to:

  * Google Workspace Admin SDK (Reports + Directory)
  * Feishu OpenAPI v3 (Contacts + Audit)
  * AWS (IAM, EC2, S3, ACM)
  * GitHub (org admin API)

Atlassian is intentionally out of scope for this PR — Helix has no
``atlassian_admin`` integration kind today; bringing it in would
require a new driver registration. The MFA / privileged review modules
fall back to per-user manual entry for Atlassian (handled by the
auditor service, not here).
"""
from __future__ import annotations

import asyncio
import json
import logging
import uuid
from collections.abc import Sequence
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import AppException
from app.integrations.models import Integration

logger = logging.getLogger(__name__)

# Canonical driver kind for software integrations. The auditor adapters
# below used to declare ``KIND = "software"``, which matches NOTHING in
# ``app.integrations.registry`` (the registry only ever knew
# ``software_provisioning``) — every ``_resolve_integration`` call from
# the Google / GitHub / Atlassian adapters raised IntegrationMissingError
# from day one and got swallowed as "integration not configured".
# ``tests/auditor/test_adapter_kinds.py`` now pins every adapter's
# ``(KIND, PROVIDER)`` to a registered driver so this cannot recur.
SOFTWARE_KIND = "software_provisioning"
CLOUD_KIND = "cloud"

# Hard cap on any paginated upstream walk. A provider that keeps
# handing back a cursor must not spin this adapter forever; hitting the
# cap logs loudly rather than pretending the page set was complete.
_MAX_PAGES = 200

# Bounded fan-out for the per-space Atlassian permission probes
# (#2188-5). Atlassian rate-limits hard, so this stays modest — the win
# is turning hundreds of SERIAL round-trips into a few waves.
_ATLASSIAN_PROBE_CONCURRENCY = 8


class IntegrationMissingError(AppException):
    """Raised when an auditor adapter is called but the upstream
    integration row is absent or disabled.

    Caller pattern:

        try:
            rows = await GoogleAuditorAdapter.list_users_mfa_status()
        except IntegrationMissingError:
            return []
    """

    def __init__(self, kind: str, provider: str | None = None) -> None:
        suffix = f"/{provider}" if provider else ""
        super().__init__(
            code="INTEGRATION_NOT_CONFIGURED",
            message=(
                f"No active integration of kind={kind}{suffix} found — "
                f"auditor read aborted; this is not a fallback, the "
                f"system genuinely has no data source for this metric."
            ),
            status_code=503,
        )


class IntegrationConfigIncompleteError(IntegrationMissingError):
    """The integration row EXISTS and is enabled, but a required config
    key is missing — a *misconfiguration*, not an absent data source.

    Both cases used to surface the same "No active integration ... found"
    sentence, which is why a single wrong key name (the Google auditor
    adapter never read ``impersonate_subject_email``) read as "we simply
    have no Google integration" in every audit row, job summary and
    triage session for the module's whole lifetime. Subclassing keeps the
    ``except IntegrationMissingError: return []`` contract intact — only
    the message changes, and it now names the keys to go fix.
    """

    def __init__(
        self,
        kind: str,
        provider: str | None = None,
        *,
        missing: Sequence[str],
    ) -> None:
        suffix = f"/{provider}" if provider else ""
        AppException.__init__(
            self,
            code="INTEGRATION_NOT_CONFIGURED",
            message=(
                f"Integration kind={kind}{suffix} is enabled but its "
                f"config is incomplete — auditor read aborted. Missing "
                f"one of: {', '.join(missing)}. This is a Helix-side "
                f"configuration gap, NOT an absent upstream data source."
            ),
            status_code=503,
        )


class _AwsAccountSkipped(Exception):
    """One AWS account could not be scanned (bad creds / API error).

    Never swallowed: the adapter either re-raises it (fail-loud default)
    or records it in the caller's ``partial`` list so the caller knows
    the round covered only a subset of the fleet.
    """

    def __init__(self, reason: str) -> None:
        self.reason = reason
        super().__init__(reason)


# ─────────────────────────────────────────────────────────────────────
# Helpers
# ─────────────────────────────────────────────────────────────────────


async def _resolve_integration(
    session: AsyncSession,
    *,
    kind: str,
    provider: str | None = None,
) -> tuple[Integration, dict[str, Any]]:
    """Return ``(integration_row, merged_config)`` for the requested
    (kind, provider). Looks up the first enabled (primary-preferred)
    integration matching, decrypts secrets via the canonical
    ``app.integrations.service`` helpers, raises
    :class:`IntegrationMissingError` if absent.
    """
    from sqlalchemy import select

    from app.integrations.service import _resolve as _resolve_one  # type: ignore

    q = select(Integration).where(
        Integration.kind == kind,
        Integration.enabled.is_(True),
    )
    if provider is not None:
        q = q.where(Integration.provider == provider)
    q = q.order_by(Integration.is_primary.desc(), Integration.created_at)
    row = (await session.execute(q)).scalars().first()
    if row is None:
        raise IntegrationMissingError(kind, provider)
    try:
        resolved = await _resolve_one(session, row)
    except AppException as exc:
        # ENCRYPTION_KEY_MISSING / disabled / etc — auditor treats
        # these as "integration unusable" and skips silently.
        raise IntegrationMissingError(kind, provider) from exc
    except Exception as exc:  # noqa: BLE001
        # cryptography.exceptions.InvalidTag (encryption key mismatch
        # between this environment and the row's secrets_nonce) — same
        # outcome: integration unusable, skip silently.
        logger.warning(
            "auditor: integration %s/%s decrypt failed (%s); "
            "treating as missing.",
            kind, provider, type(exc).__name__,
        )
        raise IntegrationMissingError(kind, provider) from exc
    return resolved.row, dict(resolved.config or {})


# ─────────────────────────────────────────────────────────────────────
# Google Workspace adapter
# ─────────────────────────────────────────────────────────────────────


@dataclass(slots=True)
class GoogleMfaRow:
    user_email: str
    google_user_id: str
    is_enrolled_in_2sv: bool
    is_enforced_in_2sv: bool
    is_admin: bool
    suspended: bool
    last_login_at: datetime | None
    raw: dict[str, Any]


@dataclass(slots=True)
class GoogleSigninRow:
    event_id: str
    user_email: str
    event_at: datetime
    event_kind: str
    status: str
    source_ip: str | None
    user_agent: str | None
    raw: dict[str, Any]


def _google_signin_rows(
    events: list[dict[str, Any]],
) -> list[GoogleSigninRow]:
    """Flatten Reports API login activities into per-sub-event rows.

    Extracted from the adapter so the id-uniqueness rule is testable
    without standing up a Google client: one activity may carry several
    sub-events (``login_challenge`` then ``login_success``), and the
    sink de-duplicates on ``(provider, provider_event_id)``. While every
    sub-event of an activity shared one id, only the first survived —
    and if the dropped one was a ``login_failure``, the
    rapid-failed-attempts detector under-counted (#2171).
    """
    rows: list[GoogleSigninRow] = []
    for raw in events:
        actor = raw.get("actor") or {}
        email = (actor.get("email") or "").lower()
        id_ = raw.get("id") or {}
        activity_id = (
            f"google:{id_.get('uniqueQualifier') or ''}:"
            f"{id_.get('time') or ''}"
        )
        event_at = _parse_iso(id_.get("time")) or datetime.now(UTC)
        for ev_index, ev in enumerate(raw.get("events") or [{}]):
            kind = ev.get("name") or "unknown"
            # Sub-event 0 KEEPS the historical id; only 1..N get a
            # suffix. Uniqueness is what #2171 needed, and this achieves
            # it without re-keying rows that are already in the ledger.
            # Suffixing every sub-event would have made the whole
            # 70-minute lookback re-ingest ONCE on deploy — and
            # ``_load_recent_failure_counts`` reads that same window, so
            # a duplicated ``login_failure`` counts twice and can trip
            # ``rapid_failed_attempts`` on a person who did nothing
            # unusual. Migration-free id changes are not free.
            event_id = (
                activity_id if ev_index == 0
                else f"{activity_id}:{ev_index}:{ev.get('name') or ''}"
            )
            params = {
                p.get("name"): p.get("value")
                for p in ev.get("parameters") or []
            }
            login_type = params.get("login_type")
            challenge = params.get("login_challenge_method")
            event_name = ev.get("name") or ""
            if event_name == "login_failure":
                status = "failure"
            elif event_name == "login_challenge":
                status = "mfa_required"
            elif event_name == "login_success":
                status = "success"
            else:
                # Unknown / unmodelled login activity — surface
                # explicitly so a future Google API change doesn't
                # silently get classified as a successful sign-in.
                status = "unknown"
            rows.append(
                GoogleSigninRow(
                    event_id=event_id,
                    user_email=email,
                    event_at=event_at,
                    event_kind=f"{kind}:{login_type or ''}",
                    status=status,
                    source_ip=raw.get("ipAddress"),
                    user_agent=params.get("user_agent"),
                    raw={"activity": raw, "event": ev,
                         "challenge": challenge},
                )
            )
    return rows


@dataclass(slots=True)
class GoogleAdminRoleRow:
    role_assignment_id: str
    role_id: str
    role_name: str
    assigned_to_user_id: str
    assigned_to_email: str | None
    assigned_at: datetime | None
    scope_type: str
    raw: dict[str, Any]


_GOOGLE_DIR_USER_RO = (
    "https://www.googleapis.com/auth/admin.directory.user.readonly"
)
_GOOGLE_DIR_ROLE_RO = (
    "https://www.googleapis.com/auth/admin.directory.rolemanagement.readonly"
)
_GOOGLE_REPORTS_RO = (
    "https://www.googleapis.com/auth/admin.reports.audit.readonly"
)


class GoogleAuditorAdapter:
    """Reads from a configured Google Workspace
    ``software_provisioning/google_workspace`` integration. Falls back
    gracefully via IntegrationMissingError.
    """

    KIND = SOFTWARE_KIND
    PROVIDER = "google_workspace"

    @classmethod
    async def list_users_mfa_status(
        cls, session: AsyncSession,
    ) -> list[GoogleMfaRow]:
        from google.oauth2 import service_account
        from googleapiclient.discovery import build

        _row, config = await _resolve_integration(
            session, kind=cls.KIND, provider=cls.PROVIDER,
        )
        sa_json = (
            config.get("service_account_json")
            or config.get("sa_json")
        )
        subject = (
            # ``impersonate_subject_email`` FIRST: it is the key the
            # canonical ``google_workspace`` driver declares in its
            # ``config_schema``, so it is the only one prod actually
            # carries. Reading only the three aliases below made every
            # Google auditor pull raise ``IntegrationMissingError`` for
            # the module's whole lifetime — MFA status, sign-in events
            # and admin-role assignments all silently returned zero rows
            # while the integration sat there enabled (2026-08-18).
            config.get("impersonate_subject_email")
            or config.get("admin_email")
            or config.get("delegated_subject")
            or config.get("subject")
        )
        if not sa_json or not subject:
            raise IntegrationConfigIncompleteError(
                cls.KIND, cls.PROVIDER,
                missing=(
                    ["service_account_json"] if not sa_json
                    else ["impersonate_subject_email"]
                ),
            )

        def _call() -> list[dict[str, Any]]:
            info = json.loads(sa_json)
            creds = service_account.Credentials.from_service_account_info(
                info, scopes=[_GOOGLE_DIR_USER_RO], subject=subject,
            )
            svc = build(
                "admin", "directory_v1", credentials=creds,
                cache_discovery=False,
            )
            users: list[dict[str, Any]] = []
            page_token: str | None = None
            while True:
                resp = svc.users().list(
                    customer="my_customer",
                    maxResults=500,
                    pageToken=page_token,
                    projection="full",
                ).execute()
                users.extend(resp.get("users") or [])
                page_token = resp.get("nextPageToken")
                if not page_token:
                    break
            return users

        users = await asyncio.to_thread(_call)
        out: list[GoogleMfaRow] = []
        for u in users:
            primary_email = (u.get("primaryEmail") or "").lower()
            if not primary_email:
                continue
            last_login_at = _parse_iso(u.get("lastLoginTime"))
            out.append(
                GoogleMfaRow(
                    user_email=primary_email,
                    google_user_id=str(u.get("id") or ""),
                    is_enrolled_in_2sv=bool(u.get("isEnrolledIn2Sv")),
                    is_enforced_in_2sv=bool(u.get("isEnforcedIn2Sv")),
                    is_admin=bool(u.get("isAdmin")),
                    suspended=bool(u.get("suspended")),
                    last_login_at=last_login_at,
                    raw=u,
                )
            )
        return out

    @classmethod
    async def list_signin_events(
        cls,
        session: AsyncSession,
        *,
        start_at: datetime,
        end_at: datetime | None = None,
    ) -> list[GoogleSigninRow]:
        """Pull login events from the Google Reports Audit API.

        Uses ``activities.list(applicationName='login')``. Time bounds
        are sent as RFC 3339 strings — the Reports API is eventually
        consistent so a 5-10 minute lag at the trailing edge is normal.
        """
        from google.oauth2 import service_account
        from googleapiclient.discovery import build

        _row, config = await _resolve_integration(
            session, kind=cls.KIND, provider=cls.PROVIDER,
        )
        sa_json = (
            config.get("service_account_json")
            or config.get("sa_json")
        )
        subject = (
            # ``impersonate_subject_email`` FIRST: it is the key the
            # canonical ``google_workspace`` driver declares in its
            # ``config_schema``, so it is the only one prod actually
            # carries. Reading only the three aliases below made every
            # Google auditor pull raise ``IntegrationMissingError`` for
            # the module's whole lifetime — MFA status, sign-in events
            # and admin-role assignments all silently returned zero rows
            # while the integration sat there enabled (2026-08-18).
            config.get("impersonate_subject_email")
            or config.get("admin_email")
            or config.get("delegated_subject")
            or config.get("subject")
        )
        if not sa_json or not subject:
            raise IntegrationConfigIncompleteError(
                cls.KIND, cls.PROVIDER,
                missing=(
                    ["service_account_json"] if not sa_json
                    else ["impersonate_subject_email"]
                ),
            )

        end_at = end_at or datetime.now(UTC)

        def _call() -> list[dict[str, Any]]:
            info = json.loads(sa_json)
            creds = service_account.Credentials.from_service_account_info(
                info, scopes=[_GOOGLE_REPORTS_RO], subject=subject,
            )
            svc = build(
                "admin", "reports_v1", credentials=creds,
                cache_discovery=False,
            )
            events: list[dict[str, Any]] = []
            page_token: str | None = None
            while True:
                resp = svc.activities().list(
                    userKey="all",
                    applicationName="login",
                    startTime=_to_rfc3339(start_at),
                    endTime=_to_rfc3339(end_at),
                    maxResults=1000,
                    pageToken=page_token,
                ).execute()
                events.extend(resp.get("items") or [])
                page_token = resp.get("nextPageToken")
                if not page_token:
                    break
            return events

        events = await asyncio.to_thread(_call)
        return _google_signin_rows(events)

    @classmethod
    async def list_admin_role_assignments(
        cls, session: AsyncSession,
    ) -> list[GoogleAdminRoleRow]:
        from google.oauth2 import service_account
        from googleapiclient.discovery import build

        _row, config = await _resolve_integration(
            session, kind=cls.KIND, provider=cls.PROVIDER,
        )
        sa_json = (
            config.get("service_account_json")
            or config.get("sa_json")
        )
        subject = (
            # ``impersonate_subject_email`` FIRST: it is the key the
            # canonical ``google_workspace`` driver declares in its
            # ``config_schema``, so it is the only one prod actually
            # carries. Reading only the three aliases below made every
            # Google auditor pull raise ``IntegrationMissingError`` for
            # the module's whole lifetime — MFA status, sign-in events
            # and admin-role assignments all silently returned zero rows
            # while the integration sat there enabled (2026-08-18).
            config.get("impersonate_subject_email")
            or config.get("admin_email")
            or config.get("delegated_subject")
            or config.get("subject")
        )
        if not sa_json or not subject:
            raise IntegrationConfigIncompleteError(
                cls.KIND, cls.PROVIDER,
                missing=(
                    ["service_account_json"] if not sa_json
                    else ["impersonate_subject_email"]
                ),
            )

        def _call() -> tuple[list[dict[str, Any]], dict[str, dict[str, Any]]]:
            info = json.loads(sa_json)
            creds = service_account.Credentials.from_service_account_info(
                info, scopes=[_GOOGLE_DIR_ROLE_RO], subject=subject,
            )
            svc = build(
                "admin", "directory_v1", credentials=creds,
                cache_discovery=False,
            )
            roles: dict[str, dict[str, Any]] = {}
            page_token: str | None = None
            while True:
                resp = svc.roles().list(
                    customer="my_customer",
                    maxResults=200,
                    pageToken=page_token,
                ).execute()
                for r in resp.get("items") or []:
                    roles[str(r.get("roleId"))] = r
                page_token = resp.get("nextPageToken")
                if not page_token:
                    break
            assigns: list[dict[str, Any]] = []
            page_token = None
            while True:
                resp = svc.roleAssignments().list(
                    customer="my_customer",
                    maxResults=200,
                    pageToken=page_token,
                ).execute()
                assigns.extend(resp.get("items") or [])
                page_token = resp.get("nextPageToken")
                if not page_token:
                    break
            return assigns, roles

        assigns, roles = await asyncio.to_thread(_call)
        admin_role_ids = {
            rid for rid, r in roles.items()
            if (r.get("isSuperAdminRole") or r.get("isSystemRole"))
            or (r.get("roleName") or "").lower().startswith("_")
        }
        # We surface ALL roleAssignments (auditor decides what's
        # privileged) but tag system roles for filter convenience.
        out: list[GoogleAdminRoleRow] = []
        for a in assigns:
            rid = str(a.get("roleId") or "")
            role_meta = roles.get(rid) or {}
            out.append(
                GoogleAdminRoleRow(
                    role_assignment_id=str(a.get("roleAssignmentId") or ""),
                    role_id=rid,
                    role_name=str(role_meta.get("roleName") or ""),
                    assigned_to_user_id=str(a.get("assignedTo") or ""),
                    assigned_to_email=None,  # only userId surfaced; UI
                                              # joins via persons table
                    assigned_at=None,
                    scope_type=str(a.get("scopeType") or ""),
                    raw={"assignment": a, "role": role_meta,
                         "is_system": rid in admin_role_ids},
                )
            )
        return out


# ─────────────────────────────────────────────────────────────────────
# AWS adapter
# ─────────────────────────────────────────────────────────────────────


@dataclass(slots=True)
class AwsSecurityGroupFinding:
    sg_id: str
    sg_name: str
    region: str
    vpc_id: str | None
    ingress_rules_open: list[dict[str, Any]] = field(default_factory=list)
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class AwsS3PublicFinding:
    bucket_name: str
    region: str | None
    public_acl: bool
    public_policy: bool
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class AwsAcmCertRow:
    arn: str
    domain_name: str
    status: str
    issuer: str | None
    not_before: datetime | None
    not_after: datetime | None
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class AwsIamAccessKeyRow:
    user_name: str
    access_key_id: str
    create_date: datetime | None
    last_used_at: datetime | None
    last_used_region: str | None
    last_used_service: str | None
    status: str
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class AwsIamRootHolder:
    account_id: str
    account_alias: str | None
    root_email: str | None
    has_mfa: bool
    has_access_keys: bool
    raw: dict[str, Any] = field(default_factory=dict)


class AwsAuditorAdapter:
    """Reads across every configured ``cloud/aws`` integration."""

    KIND = CLOUD_KIND
    PROVIDER = "aws"

    @staticmethod
    async def _list_aws_integrations(
        session: AsyncSession,
        *,
        partial: list[str] | None = None,
    ) -> list[tuple[Integration, dict[str, Any]]]:
        """Return every enabled AWS cloud integration with merged config.

        A per-account secret-resolve failure (rotated ENCRYPTION_KEY,
        InvalidTag, …) is NOT a benign skip: the caller would go on to
        scan a strict subset of the fleet and — if the caller
        auto-resolves findings it did not see this round — silently
        close every open finding belonging to the skipped account.

        ``partial`` is the opt-in channel for callers that can tolerate a
        subset **as long as they know it is a subset**: reasons are
        appended and scanning continues. Callers that pass nothing get
        the fail-loud behaviour (raise), matching the SG-scan invariant
        stated below.
        """
        from sqlalchemy import select

        from app.integrations.service import _resolve as _resolve_one  # type: ignore

        rows = (await session.execute(
            select(Integration).where(
                Integration.kind == "cloud",
                Integration.provider == "aws",
                Integration.enabled.is_(True),
            )
        )).scalars().all()
        out: list[tuple[Integration, dict[str, Any]]] = []
        for r in rows:
            try:
                resolved = await _resolve_one(session, r)
            except Exception as exc:  # noqa: BLE001
                logger.warning(
                    "AwsAuditorAdapter: resolve failed for %s: %s", r.slug, exc,
                )
                if partial is None:
                    raise
                partial.append(f"resolve_failed:{r.slug}:{exc}")
                continue
            out.append((r, dict(resolved.config or {})))
        return out

    @classmethod
    async def list_security_groups_with_open_ingress(
        cls, session: AsyncSession, *, partial: list[str] | None = None,
    ) -> list[AwsSecurityGroupFinding]:
        """Walk every region per AWS integration; return SGs with any
        0.0.0.0/0 or ::/0 ingress rule.

        This is INTENTIONALLY conservative — we surface every wide-open
        ingress rule, regardless of port. The auditor service tags
        critical ports (SSH, MySQL etc.) when persisting findings so
        the UI sorts by severity without us pre-filtering.

        ``partial`` — see :meth:`_list_aws_integrations`. Reasons land
        here for accounts that were skipped (unresolvable secrets, no
        static credentials); the caller must treat a non-empty list as
        "this was NOT a full scan" and refuse to auto-resolve.
        """
        try:
            import boto3  # noqa: F401
        except ImportError as e:  # pragma: no cover
            raise IntegrationMissingError("cloud", "aws") from e

        accounts = await cls._list_aws_integrations(session, partial=partial)
        if not accounts:
            raise IntegrationMissingError(cls.KIND, cls.PROVIDER)

        def _scan_one(config: dict[str, Any]) -> list[AwsSecurityGroupFinding]:
            ak = config.get("access_key_id") or config.get("aws_access_key_id")
            sk = config.get("secret_access_key") or config.get(
                "aws_secret_access_key",
            )
            # ``default_region`` is the ONLY region key the ``cloud/aws``
            # driver declares. The previous read (``regions`` then
            # ``region``) matched nothing an operator can actually set, so
            # every scan silently ran against the hardcoded ``us-east-1``
            # regardless of configuration — the same dead-read class as the
            # Google ``impersonate_subject_email`` miss, and a plausible
            # reason this scan reports zero findings.
            # NOTE: this honours the configured region; it is still not full
            # opted-in-region discovery (see
            # ``CmdbAwsProvider._ec2_regions_sync``). Widening a security
            # scanner to ~17 regions per account changes what it finds, not
            # just where it looks, so that is tracked separately.
            regions_cfg = [config.get("default_region") or "us-east-1"]
            if not ak or not sk:
                raise _AwsAccountSkipped("no_static_credentials")
            import boto3 as _boto3
            session_aws = _boto3.session.Session(
                aws_access_key_id=ak, aws_secret_access_key=sk,
            )
            findings: list[AwsSecurityGroupFinding] = []
            for region in regions_cfg:
                ec2 = session_aws.client("ec2", region_name=region)
                paginator = ec2.get_paginator("describe_security_groups")
                for page in paginator.paginate():
                    for sg in page.get("SecurityGroups", []):
                        open_rules: list[dict[str, Any]] = []
                        for rule in sg.get("IpPermissions") or []:
                            ranges = (
                                [r.get("CidrIp") for r in rule.get("IpRanges") or []]
                                + [r.get("CidrIpv6") for r in
                                   rule.get("Ipv6Ranges") or []]
                            )
                            if any(r in ("0.0.0.0/0", "::/0") for r in ranges):
                                open_rules.append({
                                    "from_port": rule.get("FromPort"),
                                    "to_port": rule.get("ToPort"),
                                    "ip_protocol": rule.get("IpProtocol"),
                                    "cidr_ranges": [r for r in ranges if r],
                                })
                        if not open_rules:
                            continue
                        findings.append(
                            AwsSecurityGroupFinding(
                                sg_id=sg.get("GroupId"),
                                sg_name=sg.get("GroupName"),
                                region=region,
                                vpc_id=sg.get("VpcId"),
                                ingress_rules_open=open_rules,
                                raw=sg,
                            )
                        )
            return findings

        results: list[AwsSecurityGroupFinding] = []
        for _row, config in accounts:
            try:
                results.extend(
                    await asyncio.to_thread(_scan_one, config)
                )
            except _AwsAccountSkipped as exc:
                # A configured account we cannot scan. Never silent: it
                # either aborts the scan or marks it partial.
                if partial is None:
                    raise
                logger.warning(
                    "AwsAuditorAdapter SG scan skipped %s: %s",
                    _row.slug, exc.reason,
                )
                partial.append(f"sg:{_row.slug}:{exc.reason}")
            except Exception as exc:  # noqa: BLE001
                logger.exception(
                    "AwsAuditorAdapter SG scan failed for %s: %s",
                    _row.slug, exc,
                )
                # Surface the failure to the caller — this is NOT a
                # silent fallback; we re-raise so the audit log entry
                # carries the real error.
                raise
        return results

    @classmethod
    async def list_s3_public_buckets(
        cls, session: AsyncSession, *, partial: list[str] | None = None,
    ) -> list[AwsS3PublicFinding]:
        """Public-bucket findings across every AWS integration.

        ``partial`` — see :meth:`_list_aws_integrations`. A failed
        ``list_buckets`` (rotated key, throttle, transient AccessDenied)
        used to return an empty list, which the exposure scanner read as
        "this account has no public buckets any more" and used to
        auto-resolve every open S3 finding it owned. Now such an account
        either aborts the scan (``partial is None``) or is recorded here
        so the caller skips auto-resolve.
        """
        try:
            import boto3  # noqa: F401
        except ImportError as e:  # pragma: no cover
            raise IntegrationMissingError("cloud", "aws") from e

        accounts = await cls._list_aws_integrations(session, partial=partial)
        if not accounts:
            raise IntegrationMissingError(cls.KIND, cls.PROVIDER)

        def _scan_one(
            config: dict[str, Any],
        ) -> tuple[list[AwsS3PublicFinding], list[str]]:
            reasons: list[str] = []
            ak = config.get("access_key_id") or config.get("aws_access_key_id")
            sk = config.get("secret_access_key") or config.get(
                "aws_secret_access_key",
            )
            if not ak or not sk:
                raise _AwsAccountSkipped("no_static_credentials")
            import boto3 as _boto3
            from botocore.exceptions import ClientError
            session_aws = _boto3.session.Session(
                aws_access_key_id=ak, aws_secret_access_key=sk,
            )
            s3 = session_aws.client("s3")
            findings: list[AwsS3PublicFinding] = []
            try:
                buckets = (s3.list_buckets() or {}).get("Buckets") or []
            except ClientError as exc:
                logger.warning("S3 list_buckets failed: %s", exc)
                raise _AwsAccountSkipped(f"list_buckets:{exc}") from exc
            for b in buckets:
                name = b.get("Name")
                if not name:
                    continue
                public_acl = False
                public_policy = False
                region = None
                # The three per-bucket reads (location / ACL / policy
                # status) can each fail independently — e.g. a bucket
                # with no public-access-block returns AccessDenied on
                # ``get_bucket_policy_status`` even when the ACL read
                # would have succeeded. We log the per-bucket failure
                # but continue probing the other two attributes so a
                # bucket that fails the policy check is still flagged
                # by the ACL check (and vice versa). Aborting the
                # whole bucket on any one ClientError would be a
                # silent under-count of findings.
                try:
                    loc = s3.get_bucket_location(Bucket=name) or {}
                    region = loc.get("LocationConstraint") or "us-east-1"
                except ClientError as exc:
                    logger.debug("s3 get_bucket_location %s: %s", name, exc)
                acl_read = False
                policy_read = False
                try:
                    acl = s3.get_bucket_acl(Bucket=name) or {}
                    acl_read = True
                    for grant in acl.get("Grants") or []:
                        grantee = grant.get("Grantee") or {}
                        if (
                            grantee.get("URI") == (
                                "http://acs.amazonaws.com/groups/global/AllUsers"
                            )
                            or grantee.get("URI") == (
                                "http://acs.amazonaws.com/groups/global/AuthenticatedUsers"
                            )
                        ):
                            public_acl = True
                            break
                except ClientError as exc:
                    logger.debug("s3 get_bucket_acl %s: %s", name, exc)
                try:
                    status = s3.get_bucket_policy_status(Bucket=name) or {}
                    policy_read = True
                    public_policy = bool(
                        (status.get("PolicyStatus") or {}).get("IsPublic")
                    )
                except ClientError as exc:
                    logger.debug(
                        "s3 get_bucket_policy_status %s: %s", name, exc,
                    )
                # #2231: EITHER probe failing is enough to make the round
                # partial. Findings are keyed PER ATTRIBUTE —
                # ``(AWS_S3, bucket, S3_PUBLIC_ACL)`` and
                # ``(AWS_S3, bucket, S3_PUBLIC_POLICY)`` are two separate
                # rows — but the original gate was per BUCKET and required
                # BOTH probes to fail. So a bucket whose ``get_bucket_acl``
                # returned AccessDenied (an SCP change) or SlowDown
                # (throttling — also a ClientError) while the policy probe
                # succeeded entered ``sources_scanned`` as a COMPLETE scan,
                # and ``_auto_resolve_unseen`` closed its open ACL finding
                # on the strength of a probe that errored. That is #2123's
                # mechanism reappearing one PR later, at attribute grain.
                # The comment right above already says the two probes fail
                # independently; the gate now agrees with it.
                if not acl_read:
                    reasons.append(f"bucket_probe_failed:acl:{name}")
                if not policy_read:
                    reasons.append(f"bucket_probe_failed:policy:{name}")
                if public_acl or public_policy:
                    findings.append(
                        AwsS3PublicFinding(
                            bucket_name=name,
                            region=region,
                            public_acl=public_acl,
                            public_policy=public_policy,
                            raw=b,
                        )
                    )
            return findings, reasons

        results: list[AwsS3PublicFinding] = []
        for _row, config in accounts:
            try:
                found, reasons = await asyncio.to_thread(_scan_one, config)
            except _AwsAccountSkipped as exc:
                if partial is None:
                    raise
                logger.warning(
                    "AwsAuditorAdapter S3 scan skipped %s: %s",
                    _row.slug, exc.reason,
                )
                partial.append(f"s3:{_row.slug}:{exc.reason}")
                continue
            results.extend(found)
            if reasons and partial is not None:
                partial.extend(f"s3:{_row.slug}:{r}" for r in reasons)
            elif reasons:
                raise _AwsAccountSkipped(
                    f"{_row.slug}:{','.join(reasons)}",
                )
        return results

    @classmethod
    async def list_acm_certs(
        cls, session: AsyncSession,
    ) -> list[AwsAcmCertRow]:
        try:
            import boto3  # noqa: F401
        except ImportError as e:  # pragma: no cover
            raise IntegrationMissingError("cloud", "aws") from e

        accounts = await cls._list_aws_integrations(session)
        if not accounts:
            raise IntegrationMissingError(cls.KIND, cls.PROVIDER)

        def _scan_one(config: dict[str, Any]) -> list[AwsAcmCertRow]:
            ak = config.get("access_key_id") or config.get("aws_access_key_id")
            sk = config.get("secret_access_key") or config.get(
                "aws_secret_access_key",
            )
            # ``default_region`` is the ONLY region key the ``cloud/aws``
            # driver declares. The previous read (``regions`` then
            # ``region``) matched nothing an operator can actually set, so
            # every scan silently ran against the hardcoded ``us-east-1``
            # regardless of configuration — the same dead-read class as the
            # Google ``impersonate_subject_email`` miss, and a plausible
            # reason this scan reports zero findings.
            # NOTE: this honours the configured region; it is still not full
            # opted-in-region discovery (see
            # ``CmdbAwsProvider._ec2_regions_sync``). Widening a security
            # scanner to ~17 regions per account changes what it finds, not
            # just where it looks, so that is tracked separately.
            regions_cfg = [config.get("default_region") or "us-east-1"]
            if not ak or not sk:
                return []
            import boto3 as _boto3
            session_aws = _boto3.session.Session(
                aws_access_key_id=ak, aws_secret_access_key=sk,
            )
            certs: list[AwsAcmCertRow] = []
            for region in regions_cfg:
                acm = session_aws.client("acm", region_name=region)
                paginator = acm.get_paginator("list_certificates")
                for page in paginator.paginate(
                    CertificateStatuses=["ISSUED", "PENDING_VALIDATION"],
                ):
                    for summary in page.get("CertificateSummaryList") or []:
                        arn = summary.get("CertificateArn")
                        if not arn:
                            continue
                        try:
                            detail = acm.describe_certificate(
                                CertificateArn=arn,
                            ).get("Certificate") or {}
                        except Exception as exc:  # noqa: BLE001
                            logger.warning(
                                "describe_certificate %s failed: %s",
                                arn, exc,
                            )
                            continue
                        not_before = detail.get("NotBefore")
                        not_after = detail.get("NotAfter")
                        certs.append(
                            AwsAcmCertRow(
                                arn=arn,
                                domain_name=detail.get("DomainName") or "",
                                status=detail.get("Status") or "",
                                issuer=detail.get("Issuer"),
                                not_before=_to_aware(not_before),
                                not_after=_to_aware(not_after),
                                raw=detail,
                            )
                        )
            return certs

        results: list[AwsAcmCertRow] = []
        for _row, config in accounts:
            results.extend(await asyncio.to_thread(_scan_one, config))
        return results

    @classmethod
    async def list_iam_access_keys(
        cls, session: AsyncSession,
    ) -> list[AwsIamAccessKeyRow]:
        try:
            import boto3  # noqa: F401
        except ImportError as e:  # pragma: no cover
            raise IntegrationMissingError("cloud", "aws") from e

        accounts = await cls._list_aws_integrations(session)
        if not accounts:
            raise IntegrationMissingError(cls.KIND, cls.PROVIDER)

        def _scan_one(config: dict[str, Any]) -> list[AwsIamAccessKeyRow]:
            ak = config.get("access_key_id") or config.get("aws_access_key_id")
            sk = config.get("secret_access_key") or config.get(
                "aws_secret_access_key",
            )
            if not ak or not sk:
                return []
            import boto3 as _boto3
            session_aws = _boto3.session.Session(
                aws_access_key_id=ak, aws_secret_access_key=sk,
            )
            iam = session_aws.client("iam")
            keys: list[AwsIamAccessKeyRow] = []
            paginator = iam.get_paginator("list_users")
            for page in paginator.paginate():
                for user in page.get("Users") or []:
                    user_name = user.get("UserName")
                    if not user_name:
                        continue
                    try:
                        ak_resp = iam.list_access_keys(
                            UserName=user_name,
                        ) or {}
                    except Exception as exc:  # noqa: BLE001
                        logger.warning(
                            "list_access_keys %s failed: %s",
                            user_name, exc,
                        )
                        continue
                    for k in ak_resp.get("AccessKeyMetadata") or []:
                        akid = k.get("AccessKeyId")
                        if not akid:
                            continue
                        try:
                            last = (iam.get_access_key_last_used(
                                AccessKeyId=akid,
                            ) or {}).get("AccessKeyLastUsed") or {}
                        except Exception:  # noqa: BLE001
                            last = {}
                        keys.append(
                            AwsIamAccessKeyRow(
                                user_name=user_name,
                                access_key_id=akid,
                                create_date=_to_aware(k.get("CreateDate")),
                                last_used_at=_to_aware(last.get("LastUsedDate")),
                                last_used_region=last.get("Region"),
                                last_used_service=last.get("ServiceName"),
                                status=k.get("Status") or "",
                                raw={"key": k, "last_used": last,
                                     "user": user},
                            )
                        )
            return keys

        results: list[AwsIamAccessKeyRow] = []
        for _row, config in accounts:
            results.extend(await asyncio.to_thread(_scan_one, config))
        return results

    @classmethod
    async def list_root_account_holders(
        cls, session: AsyncSession,
    ) -> list[AwsIamRootHolder]:
        """Best-effort: enumerate AWS account aliases + summary so the
        privileged-review module can show 'who has root keys' per
        account.

        Real AWS root credentials are NOT enumerable via API for the
        same account (you can only check ``GetAccountSummary`` for the
        existence of root-account access keys / MFA). We return one
        synthetic row per integration with that summary.
        """
        try:
            import boto3  # noqa: F401
        except ImportError as e:  # pragma: no cover
            raise IntegrationMissingError("cloud", "aws") from e

        accounts = await cls._list_aws_integrations(session)
        if not accounts:
            raise IntegrationMissingError(cls.KIND, cls.PROVIDER)

        def _scan_one(config: dict[str, Any]) -> AwsIamRootHolder | None:
            ak = config.get("access_key_id") or config.get("aws_access_key_id")
            sk = config.get("secret_access_key") or config.get(
                "aws_secret_access_key",
            )
            if not ak or not sk:
                return None
            import boto3 as _boto3
            session_aws = _boto3.session.Session(
                aws_access_key_id=ak, aws_secret_access_key=sk,
            )
            sts = session_aws.client("sts")
            iam = session_aws.client("iam")
            try:
                ident = sts.get_caller_identity() or {}
            except Exception:  # noqa: BLE001
                return None
            try:
                aliases = (iam.list_account_aliases() or {}).get(
                    "AccountAliases"
                ) or []
            except Exception:  # noqa: BLE001
                aliases = []
            try:
                summary = (iam.get_account_summary() or {}).get(
                    "SummaryMap"
                ) or {}
            except Exception:  # noqa: BLE001
                summary = {}
            return AwsIamRootHolder(
                account_id=str(ident.get("Account") or ""),
                account_alias=aliases[0] if aliases else None,
                root_email=None,  # not enumerable via API
                has_mfa=bool(summary.get("AccountMFAEnabled") == 1),
                has_access_keys=bool(summary.get("AccountAccessKeysPresent") and
                                     summary.get("AccountAccessKeysPresent") > 0),
                raw={"identity": ident, "aliases": aliases,
                     "summary": summary},
            )

        results: list[AwsIamRootHolder] = []
        for _row, config in accounts:
            r = await asyncio.to_thread(_scan_one, config)
            if r is not None:
                results.append(r)
        return results


# ─────────────────────────────────────────────────────────────────────
# GitHub adapter
# ─────────────────────────────────────────────────────────────────────


@dataclass(slots=True)
class GitHubPatRow:
    token_id: str
    owner_login: str
    owner_email: str | None
    name: str | None
    scopes: list[str]
    created_at: datetime | None
    last_used_at: datetime | None
    expires_at: datetime | None
    is_revoked: bool
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class GitHubAdminRow:
    user_login: str
    user_email: str | None
    role: str
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class GitHubMemberMfaRow:
    user_login: str
    user_email: str | None
    is_2fa_enabled: bool
    role: str  # "admin" or "member"
    raw: dict[str, Any] = field(default_factory=dict)


class GitHubAuditorAdapter:
    KIND = SOFTWARE_KIND
    PROVIDER = "github"

    @classmethod
    async def list_organization_pats(
        cls, session: AsyncSession,
    ) -> list[GitHubPatRow]:
        """List PATs / fine-grained PATs across the configured org via
        GitHub Enterprise Cloud's
        ``/orgs/{org}/personal-access-tokens`` endpoint.

        Requires the integration's PAT/App to have
        ``Personal access tokens (read)`` permission at the org level.
        """
        import httpx

        _row, config = await _resolve_integration(
            session, kind=cls.KIND, provider=cls.PROVIDER,
        )
        # ``org_slug`` FIRST — the only key the ``github`` driver declares.
        # The two aliases below match nothing in prod (same class of bug as
        # the Google ``impersonate_subject_email`` miss); they stay as
        # tolerated aliases, not as the primary read.
        org = (
            config.get("org_slug")
            or config.get("organization")
            or config.get("org")
        )
        token = (
            config.get("admin_token")
            or config.get("api_token")
            or config.get("token")
        )
        if not org or not token:
            raise IntegrationMissingError(cls.KIND, cls.PROVIDER)

        headers = {
            "Accept": "application/vnd.github+json",
            "Authorization": f"Bearer {token}",
            "X-GitHub-Api-Version": "2022-11-28",
        }
        url = (
            f"https://api.github.com/orgs/{org}/personal-access-tokens"
            f"?per_page=100"
        )
        rows: list[GitHubPatRow] = []
        async with httpx.AsyncClient(timeout=30.0, trust_env=False) as client:
            while url:
                resp = await client.get(url, headers=headers)
                if resp.status_code == 404:
                    # Endpoint requires Enterprise plan w/ PAT policy
                    # configured. Surface as missing rather than empty
                    # list — the auditor needs to know.
                    raise IntegrationMissingError(
                        f"{cls.KIND} ({cls.PROVIDER}: pat_listing_unavailable)",
                    )
                resp.raise_for_status()
                for r in resp.json() or []:
                    owner = r.get("owner") or {}
                    rows.append(
                        GitHubPatRow(
                            token_id=str(r.get("id")),
                            owner_login=owner.get("login") or "",
                            owner_email=None,
                            name=r.get("token_name"),
                            scopes=list(
                                r.get("permissions", {}).get("repository", {})
                                .keys()
                            )
                            + list(
                                r.get("permissions", {}).get("organization", {})
                                .keys()
                            ),
                            created_at=_parse_iso(r.get("created_at")),
                            last_used_at=_parse_iso(r.get("last_used_at")),
                            expires_at=_parse_iso(r.get("expires_at")),
                            is_revoked=False,
                            raw=r,
                        )
                    )
                # GitHub uses Link header for pagination
                link = resp.headers.get("Link", "")
                next_url = None
                for part in link.split(","):
                    if 'rel="next"' in part:
                        seg = part.split(";")[0].strip()
                        next_url = seg.lstrip("<").rstrip(">")
                        break
                url = next_url
        return rows

    @classmethod
    async def list_organization_admins(
        cls, session: AsyncSession,
    ) -> list[GitHubAdminRow]:
        import httpx

        _row, config = await _resolve_integration(
            session, kind=cls.KIND, provider=cls.PROVIDER,
        )
        # ``org_slug`` FIRST — the only key the ``github`` driver declares.
        # The two aliases below match nothing in prod (same class of bug as
        # the Google ``impersonate_subject_email`` miss); they stay as
        # tolerated aliases, not as the primary read.
        org = (
            config.get("org_slug")
            or config.get("organization")
            or config.get("org")
        )
        token = (
            config.get("admin_token")
            or config.get("api_token")
            or config.get("token")
        )
        if not org or not token:
            raise IntegrationMissingError(cls.KIND, cls.PROVIDER)
        headers = {
            "Accept": "application/vnd.github+json",
            "Authorization": f"Bearer {token}",
            "X-GitHub-Api-Version": "2022-11-28",
        }
        url = (
            f"https://api.github.com/orgs/{org}/members"
            f"?role=admin&per_page=100"
        )
        rows: list[GitHubAdminRow] = []
        async with httpx.AsyncClient(timeout=30.0, trust_env=False) as client:
            while url:
                resp = await client.get(url, headers=headers)
                resp.raise_for_status()
                for u in resp.json() or []:
                    rows.append(
                        GitHubAdminRow(
                            user_login=u.get("login") or "",
                            user_email=None,
                            role="admin",
                            raw=u,
                        )
                    )
                link = resp.headers.get("Link", "")
                next_url = None
                for part in link.split(","):
                    if 'rel="next"' in part:
                        seg = part.split(";")[0].strip()
                        next_url = seg.lstrip("<").rstrip(">")
                        break
                url = next_url
        return rows

    @classmethod
    async def list_organization_members_with_mfa(
        cls, session: AsyncSession,
    ) -> list[GitHubMemberMfaRow]:
        """Return one row per org member with 2FA enrollment status.

        Strategy:
          1. Page ``/orgs/{org}/members?per_page=100`` → full member set
             (login + role).
          2. Page ``/orgs/{org}/members?filter=2fa_disabled&per_page=100``
             → set of logins WITHOUT 2FA enabled.
          3. Set difference → ``is_2fa_enabled`` per member.

        Step 2 requires the caller to be an org owner; GitHub silently
        returns the same list as step 1 if the caller lacks owner role
        (which would mark every member as MFA-disabled). To guard
        against that false positive, when the disabled set ⊇ the full
        set, we treat the answer as "policy not visible" and raise
        :class:`IntegrationMissingError` — the auditor sees an explicit
        "GitHub PAT lacks org-owner scope, MFA data unavailable" error
        rather than a fleet-wide false alarm.
        """
        import httpx

        _row, config = await _resolve_integration(
            session, kind=cls.KIND, provider=cls.PROVIDER,
        )
        # ``org_slug`` FIRST — the only key the ``github`` driver declares.
        # The two aliases below match nothing in prod (same class of bug as
        # the Google ``impersonate_subject_email`` miss); they stay as
        # tolerated aliases, not as the primary read.
        org = (
            config.get("org_slug")
            or config.get("organization")
            or config.get("org")
        )
        token = (
            config.get("admin_token")
            or config.get("api_token")
            or config.get("token")
        )
        if not org or not token:
            raise IntegrationMissingError(cls.KIND, cls.PROVIDER)

        headers = {
            "Accept": "application/vnd.github+json",
            "Authorization": f"Bearer {token}",
            "X-GitHub-Api-Version": "2022-11-28",
        }

        async def _paginate(initial_url: str) -> list[dict[str, Any]]:
            out: list[dict[str, Any]] = []
            url: str | None = initial_url
            async with httpx.AsyncClient(timeout=30.0, trust_env=False) as client:
                while url:
                    resp = await client.get(url, headers=headers)
                    resp.raise_for_status()
                    out.extend(resp.json() or [])
                    link = resp.headers.get("Link", "")
                    next_url = None
                    for part in link.split(","):
                        if 'rel="next"' in part:
                            seg = part.split(";")[0].strip()
                            next_url = seg.lstrip("<").rstrip(">")
                            break
                    url = next_url
            return out

        all_members = await _paginate(
            f"https://api.github.com/orgs/{org}/members?per_page=100"
        )
        admin_members = await _paginate(
            f"https://api.github.com/orgs/{org}/members"
            f"?role=admin&per_page=100"
        )
        admin_logins = {(m.get("login") or "").lower() for m in admin_members}

        try:
            disabled_members = await _paginate(
                f"https://api.github.com/orgs/{org}/members"
                f"?filter=2fa_disabled&per_page=100"
            )
        except httpx.HTTPStatusError as exc:
            # 422 means filter unsupported on this plan — treat as no
            # MFA data; raise so caller doesn't false-flag everyone.
            if exc.response.status_code in (403, 422):
                raise IntegrationMissingError(
                    f"{cls.KIND} ({cls.PROVIDER}: 2fa_filter_unavailable)"
                ) from exc
            raise

        disabled_logins = {
            (m.get("login") or "").lower() for m in disabled_members
        }

        # Guard: if the "disabled" set isn't strictly a subset of the
        # full set we got back, the PAT probably lacks org-owner scope
        # and GitHub returned the unfiltered list both times — refuse to
        # write. Use ⊆ check (every disabled login must be a known
        # member); if disabled_logins has unknown logins, something is
        # wrong with our pagination — also refuse.
        all_logins = {(m.get("login") or "").lower() for m in all_members}
        if not disabled_logins.issubset(all_logins):
            raise IntegrationMissingError(
                f"{cls.KIND} ({cls.PROVIDER}: 2fa_filter_returned_unexpected)"
            )
        # Heuristic guard: if every member is reported as disabled, the
        # filter likely wasn't applied (PAT scope issue). Real orgs
        # almost never have 100% MFA-off — admins enforce it.
        if all_logins and disabled_logins == all_logins:
            raise IntegrationMissingError(
                f"{cls.KIND} ({cls.PROVIDER}: 2fa_filter_likely_unapplied)"
            )

        rows: list[GitHubMemberMfaRow] = []
        for m in all_members:
            login = (m.get("login") or "").lower()
            if not login:
                continue
            rows.append(
                GitHubMemberMfaRow(
                    user_login=login,
                    user_email=None,
                    is_2fa_enabled=login not in disabled_logins,
                    role="admin" if login in admin_logins else "member",
                    raw=m,
                )
            )
        return rows


# ─────────────────────────────────────────────────────────────────────
# Feishu adapter
# ─────────────────────────────────────────────────────────────────────


def _feishu_email(user: dict[str, Any]) -> str | None:
    """Corporate email for a Feishu contact row.

    ``enterprise_email`` first, personal ``email`` only as a fallback —
    the same precedence every other Feishu consumer in this repo uses
    (``attendance/feishu_attendance.py``, ``identity_sync/service.py``,
    ``auth/service.py``, the last one carrying a 2026-05-29 production
    incident note). Reading ``email`` alone left every user who binds
    only a corporate address unmatched, so their MFA state never
    reached the ledger and the coverage ratio silently under-counted.
    """
    for key in ("enterprise_email", "email"):
        value = (user.get(key) or "").strip()
        if value:
            return value
    return None


@dataclass(slots=True)
class FeishuMfaRow:
    user_email: str | None
    open_id: str
    employee_no: str | None
    is_2step_enabled: bool
    # False when the tenant did not return ``security_strategy`` at all
    # (the app lacks enterprise + audit scope): then ``is_2step_enabled``
    # is this dataclass's default, not something Feishu told us. Consumers
    # must not score an unobserved user as non-compliant.
    is_2step_observed: bool = False
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class FeishuAdminRoleRow:
    user_id: str
    user_email: str | None
    role_name: str
    raw: dict[str, Any] = field(default_factory=dict)


class FeishuAuditorAdapter:
    """Reads from the configured ``hris/successfactors`` or
    ``chat_approval/feishu`` integration credentials.

    Feishu's MFA / 2-step status is exposed under the user's
    ``security_strategy`` flag in their Contact API — accessible to a
    Feishu app with ``contact:user.base:readonly`` permission.
    """

    KIND_CHAT = "chat_approval"
    PROVIDER_CHAT = "feishu"

    @classmethod
    async def _tenant_token(
        cls, session: AsyncSession,
    ) -> tuple[str, dict[str, Any]]:
        import httpx

        _row, config = await _resolve_integration(
            session, kind=cls.KIND_CHAT, provider=cls.PROVIDER_CHAT,
        )
        app_id = config.get("app_id")
        app_secret = config.get("app_secret")
        if not app_id or not app_secret:
            raise IntegrationMissingError(cls.KIND_CHAT, cls.PROVIDER_CHAT)
        async with httpx.AsyncClient(timeout=15.0, trust_env=False) as client:
            resp = await client.post(
                "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
                json={"app_id": app_id, "app_secret": app_secret},
            )
            resp.raise_for_status()
            data = resp.json() or {}
            token = data.get("tenant_access_token")
            if not token:
                raise IntegrationMissingError(
                    cls.KIND_CHAT, cls.PROVIDER_CHAT,
                )
            return token, config

    @classmethod
    async def list_users_mfa_status(
        cls, session: AsyncSession,
    ) -> list[FeishuMfaRow]:
        """Enumerate Feishu users with their 2-step status.

        Strategy: most tenants do NOT grant the app full-tenant
        ``department_id=0`` scope (that path returns 400 Bad Request),
        so we go through ``/contact/v3/scopes`` first to find which
        departments + explicit users the app can read, then iterate.

        Surface every reachable user; ``security_strategy`` is null
        for tenants without enterprise + audit scope, in which case we
        store ``is_2step_enabled=False`` and the auditor flags it as
        missing.
        """
        import httpx

        token, _ = await cls._tenant_token(session)
        headers = {"Authorization": f"Bearer {token}"}
        out: list[FeishuMfaRow] = []
        seen_open_ids: set[str] = set()

        async def _fetch_dept(
            client: Any, dept_id: str,
        ) -> None:
            page_token: str | None = None
            while True:
                params: dict[str, Any] = {
                    "page_size": 100,
                    "department_id": dept_id,
                    "user_id_type": "open_id",
                }
                if page_token:
                    params["page_token"] = page_token
                resp = await client.get(
                    "https://open.feishu.cn/open-apis/"
                    "contact/v3/users/find_by_department",
                    headers=headers, params=params,
                )
                if resp.status_code in (400, 403, 404):
                    # Per-department permission gap — skip rather than
                    # 500 the whole sync. Other depts may still be
                    # reachable.
                    return
                resp.raise_for_status()
                data = (resp.json() or {}).get("data") or {}
                for u in data.get("items") or []:
                    open_id = u.get("open_id") or ""
                    if open_id and open_id in seen_open_ids:
                        continue
                    if open_id:
                        seen_open_ids.add(open_id)
                    out.append(
                        FeishuMfaRow(
                            user_email=_feishu_email(u),
                            open_id=open_id,
                            employee_no=u.get("employee_no"),
                            is_2step_enabled=bool(
                                (u.get("security_strategy") or {}).get(
                                    "two_factor_enabled"
                                )
                            ),
                            raw=u,
                        )
                    )
                if not data.get("has_more"):
                    break
                page_token = data.get("page_token")
                if not page_token:
                    break

        async def _fetch_user(
            client: Any, user_open_id: str,
        ) -> None:
            if user_open_id in seen_open_ids:
                return
            resp = await client.get(
                f"https://open.feishu.cn/open-apis/"
                f"contact/v3/users/{user_open_id}",
                headers=headers,
                params={"user_id_type": "open_id"},
            )
            if resp.status_code in (400, 403, 404):
                return
            resp.raise_for_status()
            u = (resp.json() or {}).get("data", {}).get("user") or {}
            if not u:
                return
            open_id = u.get("open_id") or user_open_id
            if open_id in seen_open_ids:
                return
            seen_open_ids.add(open_id)
            out.append(
                FeishuMfaRow(
                    user_email=_feishu_email(u),
                    open_id=open_id,
                    employee_no=u.get("employee_no"),
                    is_2step_enabled=bool(
                        (u.get("security_strategy") or {}).get(
                            "two_factor_enabled"
                        )
                    ),
                    is_2step_observed=(
                        u.get("security_strategy") is not None
                    ),
                    raw=u,
                )
            )

        async with httpx.AsyncClient(timeout=30.0, trust_env=False) as client:
            # 1. Discover what scope the Feishu app has.
            scope_resp = await client.get(
                "https://open.feishu.cn/open-apis/contact/v3/scopes",
                headers=headers, params={"user_id_type": "open_id"},
            )
            dept_ids: list[str] = []
            user_ids: list[str] = []
            if scope_resp.status_code == 200:
                scope_data = (scope_resp.json() or {}).get("data") or {}
                dept_ids = list(scope_data.get("department_ids") or [])
                user_ids = list(scope_data.get("user_ids") or [])

            # 2. If no scoped departments, the app has full-tenant
            #    visibility — root department_id "0" is the standard
            #    Feishu sentinel for "the whole org". Try it; if it
            #    400s the per-dept handler skips silently (and we fall
            #    back to user_ids only).
            if not dept_ids:
                dept_ids = ["0"]

            # 3. Iterate departments, dedupe by open_id.
            for dept_id in dept_ids:
                await _fetch_dept(client, dept_id)

            # 4. Pick up users explicitly granted to the app outside
            #    those departments.
            for uid in user_ids:
                await _fetch_user(client, uid)

        return out

    @classmethod
    async def list_admin_role_holders(
        cls, session: AsyncSession,
    ) -> list[FeishuAdminRoleRow]:
        """Feishu admin / superadmin enumeration goes through
        ``admin.v1.password_admins`` (only available with admin scope).
        Returns empty list silently when scope absent — the auditor
        flags missing data.
        """
        import httpx

        token, _ = await cls._tenant_token(session)
        headers = {"Authorization": f"Bearer {token}"}
        out: list[FeishuAdminRoleRow] = []
        try:
            async with httpx.AsyncClient(timeout=15.0, trust_env=False) as client:
                resp = await client.get(
                    "https://open.feishu.cn/open-apis/admin/v1/admin_role_assignments",
                    headers=headers,
                )
                if resp.status_code in (403, 404):
                    return out
                resp.raise_for_status()
                for r in (resp.json() or {}).get("data", {}).get(
                    "items"
                ) or []:
                    out.append(
                        FeishuAdminRoleRow(
                            user_id=r.get("user_id") or "",
                            user_email=_feishu_email(r),
                            role_name=r.get("role_name") or "admin",
                            raw=r,
                        )
                    )
        except httpx.HTTPStatusError as e:
            if e.response.status_code in (403, 404):
                return out
            raise
        return out


# ─────────────────────────────────────────────────────────────────────
# Helpers
# ─────────────────────────────────────────────────────────────────────


def _parse_iso(value: Any) -> datetime | None:
    if not value:
        return None
    if isinstance(value, datetime):
        return _to_aware(value)
    s = str(value).strip()
    if not s:
        return None
    try:
        if s.endswith("Z"):
            s = s[:-1] + "+00:00"
        d = datetime.fromisoformat(s)
        return _to_aware(d)
    except (TypeError, ValueError):
        return None


def _to_aware(d: datetime | None) -> datetime | None:
    if d is None:
        return None
    if d.tzinfo is None:
        return d.replace(tzinfo=UTC)
    return d


def _to_rfc3339(d: datetime) -> str:
    if d.tzinfo is None:
        d = d.replace(tzinfo=UTC)
    return d.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%S.000Z")


# ═════════════════════════════════════════════════════════════════════
# Atlassian (re-uses ``software/jira`` integration credentials)
# ═════════════════════════════════════════════════════════════════════
#
# Credentials live on the ``software/jira`` integration row already —
# Helix has no separate ``atlassian_admin`` driver. JiraSoftwareDriver's
# config_schema declares both ``admin_email`` + ``api_token`` (Basic, for
# /rest/api/3/) AND ``org_id`` + ``org_admin_api_key`` (Bearer, for
# /admin/v1/orgs/{org_id}/). The auditor methods below pull from the
# same row.
#
# When ``org_id`` / ``org_admin_api_key`` are absent (Atlassian Free /
# Standard plan, or Guard not subscribed), only the site-level methods
# work; calls that need /admin/v1/ raise IntegrationMissingError with a
# message naming the missing field, so the auditor surfaces "Atlassian
# Guard not configured" instead of silently returning empty data.


_ATLASSIAN_ADMIN_BASE = "https://api.atlassian.com/admin/v1"


@dataclass(slots=True)
class AtlassianMfaPolicyRow:
    """Atlassian exposes 2-step verification status as an *org policy*,
    not a per-user attribute. We surface one row per (user, policy) so
    the MFA dashboard can compare each user against the policy that
    governs their directory.
    """
    user_email: str | None
    user_account_id: str
    is_enrolled: bool   # derived from policy state for the user's directory
    policy_status: str  # "enforced" | "not_enforced" | "policy_absent"
    policy_directory_id: str | None
    last_synced_at: datetime
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class AtlassianAdminRow:
    user_account_id: str
    user_email: str | None
    user_name: str | None
    is_org_admin: bool
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class AtlassianPublicSpaceRow:
    site_url: str
    space_key: str
    space_name: str | None
    space_type: str  # "confluence_space" | "jira_project"
    public_url: str | None
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class AtlassianApiTokenRow:
    token_id: str
    user_account_id: str
    user_email: str | None
    label: str | None
    created_at: datetime | None
    last_used_at: datetime | None
    raw: dict[str, Any] = field(default_factory=dict)


class AtlassianAuditorAdapter:
    """Auditor-side reads against Atlassian Cloud Org Admin API.

    Re-uses the ``software/jira`` integration row — the same one that
    drives Jira/Confluence seat provisioning. Three of the four methods
    require ``org_admin_api_key`` (Atlassian Guard subscription); they
    raise IntegrationMissingError otherwise so the auditor surfaces
    the real configuration gap instead of empty data.
    """

    KIND = SOFTWARE_KIND
    PROVIDER = "jira"

    @classmethod
    async def _resolve_org_admin_creds(
        cls, session: AsyncSession,
    ) -> tuple[str, str]:
        """Return ``(org_id, bearer_token)`` for /admin/v1/ access.

        Raises IntegrationMissingError if either is missing. Atlassian
        Guard / Atlassian Access are paid add-ons that gate the Org
        Admin API; absence is a *real* state — we surface it as a
        config gap rather than an empty data set.
        """
        _row, config = await _resolve_integration(
            session, kind=cls.KIND, provider=cls.PROVIDER,
        )
        org_id = (config.get("org_id") or "").strip()
        org_key = (config.get("org_admin_api_key") or "").strip()
        if not org_id:
            raise IntegrationMissingError(
                f"{cls.KIND} ({cls.PROVIDER}: org_id missing — "
                f"Atlassian Org ID required for /admin/v1/ calls)"
            )
        if not org_key:
            raise IntegrationMissingError(
                f"{cls.KIND} ({cls.PROVIDER}: org_admin_api_key missing — "
                f"generate at admin.atlassian.com/o/{{orgId}}/api-keys)"
            )
        return org_id, org_key

    @classmethod
    def _admin_headers(cls, bearer: str) -> dict[str, str]:
        return {
            "Authorization": f"Bearer {bearer}",
            "Accept": "application/json",
        }

    @classmethod
    async def _paginate_admin(
        cls,
        client: Any,  # httpx.AsyncClient — imported lazily per-method
        url: str,
        headers: dict[str, str],
    ) -> list[dict[str, Any]]:
        """Walk Atlassian's cursor pagination — `data` + `links.next`."""
        out: list[dict[str, Any]] = []
        next_url: str | None = url
        while next_url:
            resp = await client.get(next_url, headers=headers)
            resp.raise_for_status()
            body = resp.json() or {}
            out.extend(body.get("data") or [])
            links = body.get("links") or {}
            next_url = links.get("next")
        return out

    @classmethod
    async def list_users_mfa_status(
        cls, session: AsyncSession,
    ) -> list[AtlassianMfaPolicyRow]:
        """Per-user MFA derived from the org's 2-step-verification policy.

        Atlassian does NOT expose per-user 2FA enrollment on its public
        API — even Atlassian Guard / Access don't surface a
        ``has_two_factor`` flag. The supported pattern is to read the
        org's authentication policies (``/admin/v1/orgs/{org}/policies``,
        ``type=authentication-policy``) and check whether
        ``two-step-verification`` is required for the user's directory.
        We map that policy state to ``is_enrolled``:

          * ``enforced``     → ``is_enrolled=True`` (user MUST 2FA to log in)
          * ``not_enforced`` → ``is_enrolled=False``
          * ``policy_absent``→ ``is_enrolled=False`` + raw_payload note

        The result is honest: when the org has 2FA strictly enforced,
        every active user is compliant; when not, we say so. We never
        fabricate per-user enrollment we cannot observe.
        """
        import httpx

        org_id, bearer = await cls._resolve_org_admin_creds(session)
        headers = cls._admin_headers(bearer)
        synced_at = datetime.now(UTC)

        async with httpx.AsyncClient(timeout=30.0, trust_env=False) as client:
            policies_resp = await client.get(
                f"{_ATLASSIAN_ADMIN_BASE}/orgs/{org_id}/policies"
                f"?type=authentication-policy",
                headers=headers,
            )
            policies_resp.raise_for_status()
            policies = (policies_resp.json() or {}).get("data") or []

            # Index 2FA policy state per directory it covers. The
            # ``attributes.rule`` for two-step verification is
            # ``two-step-verification`` with ``status`` enabled / disabled.
            policy_per_dir: dict[str, str] = {}
            for p in policies:
                attrs = p.get("attributes") or {}
                if not attrs.get("status") == "enabled":
                    continue
                rules = (attrs.get("rule") or [])
                if "two-step-verification" not in rules:
                    continue
                # Each policy lists directories it applies to.
                for d in (p.get("relationships", {})
                          .get("directories", {}).get("data") or []):
                    dir_id = d.get("id")
                    if dir_id:
                        # If multiple policies cover the same directory,
                        # the strictest wins — "enforced" trumps anything.
                        policy_per_dir[dir_id] = "enforced"

            # Walk users — Atlassian's user list endpoint is per-org
            # under /admin/v1/orgs/{orgId}/users.
            users = await cls._paginate_admin(
                client,
                f"{_ATLASSIAN_ADMIN_BASE}/orgs/{org_id}/users",
                headers,
            )

        out: list[AtlassianMfaPolicyRow] = []
        for u in users:
            attrs = u.get("attributes") or u  # API shape varies
            account_id = (
                u.get("id") or attrs.get("account_id") or ""
            )
            email = (
                attrs.get("email") or attrs.get("emailAddress") or None
            )
            # Resolve the user's directory — uses ``account_status`` /
            # ``directory_id`` field; absence means we cannot evaluate.
            dir_id = (
                attrs.get("directory_id")
                or (attrs.get("attributes") or {}).get("directory_id")
            )
            if dir_id and dir_id in policy_per_dir:
                policy_status = "enforced"
                is_enrolled = True
            elif policy_per_dir:
                policy_status = "not_enforced"
                is_enrolled = False
            else:
                policy_status = "policy_absent"
                is_enrolled = False
            out.append(AtlassianMfaPolicyRow(
                user_email=(email.lower() if email else None),
                user_account_id=str(account_id),
                is_enrolled=is_enrolled,
                policy_status=policy_status,
                policy_directory_id=dir_id,
                last_synced_at=synced_at,
                raw=u,
            ))
        return out

    @classmethod
    async def list_organization_admins(
        cls, session: AsyncSession,
    ) -> list[AtlassianAdminRow]:
        """Org-level admins via ``/admin/v1/orgs/{orgId}/users`` filtered
        by ``account_type=managed`` + ``role=org-admin``. The role list
        Atlassian returns is open-ended; we surface only the
        ``org-admin`` and ``site-admin`` rows.
        """
        import httpx

        org_id, bearer = await cls._resolve_org_admin_creds(session)
        headers = cls._admin_headers(bearer)

        async with httpx.AsyncClient(timeout=30.0, trust_env=False) as client:
            # The "org roles" endpoint is the canonical source for
            # Atlassian Org Admin role assignments. Falls under
            # /admin/v1/orgs/{orgId}/users with role filter.
            users = await cls._paginate_admin(
                client,
                f"{_ATLASSIAN_ADMIN_BASE}/orgs/{org_id}/users",
                headers,
            )

        out: list[AtlassianAdminRow] = []
        for u in users:
            attrs = u.get("attributes") or u
            roles = (attrs.get("roles") or [])
            is_admin = any(
                r in ("org-admin", "organization-admin", "site-admin")
                for r in roles
            )
            if not is_admin:
                continue
            account_id = u.get("id") or attrs.get("account_id") or ""
            email = (
                attrs.get("email")
                or attrs.get("emailAddress")
                or None
            )
            out.append(AtlassianAdminRow(
                user_account_id=str(account_id),
                user_email=(email.lower() if email else None),
                user_name=attrs.get("name") or attrs.get("displayName"),
                is_org_admin=True,
                raw=u,
            ))
        return out

    @classmethod
    async def list_public_spaces(
        cls, session: AsyncSession,
    ) -> list[AtlassianPublicSpaceRow]:
        """Public/anonymous-accessible Confluence spaces + Jira projects.

        Uses the *site-level* REST API (Basic auth, admin_email +
        api_token). Confluence: ``/wiki/rest/api/space?type=global`` then
        per-space ``/wiki/rest/api/space/{key}/permission`` to find
        anonymous "view" permission. Jira: ``/rest/api/3/project/search``
        with ``expand=permissions`` to find anonymous role assignments.

        The site_url + admin_email + api_token must be present (Helix
        always requires these for the Jira driver to function). If
        Confluence isn't enabled on the site the Confluence call
        returns 404 and we treat it as "no Confluence findings"
        (genuinely no data).
        """
        import httpx

        _row, config = await _resolve_integration(
            session, kind=cls.KIND, provider=cls.PROVIDER,
        )
        site_url = (config.get("site_url") or "").strip()
        admin_email = (config.get("admin_email") or "").strip()
        api_token = (config.get("api_token") or "").strip()
        if not (site_url and admin_email and api_token):
            raise IntegrationMissingError(cls.KIND, cls.PROVIDER)

        import base64
        auth_header = (
            "Basic "
            + base64.b64encode(
                f"{admin_email}:{api_token}".encode()
            ).decode()
        )
        site_base = f"https://{site_url}"
        headers = {
            "Authorization": auth_header,
            "Accept": "application/json",
        }

        out: list[AtlassianPublicSpaceRow] = []
        async with httpx.AsyncClient(timeout=30.0, trust_env=False) as client:
            # ── Confluence anonymous spaces ─────────────────────
            try:
                start = 0
                while True:
                    resp = await client.get(
                        f"{site_base}/wiki/rest/api/space"
                        f"?type=global&start={start}&limit=100",
                        headers=headers,
                    )
                    if resp.status_code == 404:
                        break  # Confluence not enabled
                    resp.raise_for_status()
                    body = resp.json() or {}
                    spaces = body.get("results") or []
                    # Probe the page's spaces CONCURRENTLY (#2188-5).
                    # One permission call per space, serially, is a few
                    # hundred sequential round-trips on a real
                    # Confluence site — minutes of wall clock for a scan
                    # that is entirely I/O-bound. Bounded fan-out keeps
                    # it fast without hammering the tenant (Atlassian
                    # rate-limits aggressively).
                    probe_sem = asyncio.Semaphore(_ATLASSIAN_PROBE_CONCURRENCY)

                    async def _probe(space: dict[str, Any]) -> Any:
                        space_key = space.get("key")
                        if not space_key:
                            return None
                        async with probe_sem:
                            return space, await client.get(
                                f"{site_base}/wiki/rest/api/space/"
                                f"{space_key}/permission",
                                headers=headers,
                            )

                    probed = await asyncio.gather(
                        *[_probe(sp) for sp in spaces],
                        return_exceptions=True,
                    )
                    for probe_result in probed:
                        if isinstance(probe_result, BaseException):
                            # A single space's probe failing must not
                            # abort the page. Unlike the serial version
                            # (where an exception propagated) this one
                            # swallows it, so it has to SAY so — an
                            # unlogged skip is how a systematically
                            # failing probe reads as "no public spaces".
                            logger.warning(
                                "Atlassian space permission probe failed: "
                                "%r", probe_result,
                            )
                            continue
                        if probe_result is None:
                            # Space carried no key — nothing to probe.
                            continue
                        sp, perm = probe_result
                        key = sp.get("key")
                        if perm.status_code != 200:
                            continue
                        perms = perm.json() or {}
                        is_anon = any(
                            (p.get("subject") or {}).get("type")
                                == "anonymous"
                            and p.get("operation", {}).get(
                                "operation"
                            ) in ("read", "view")
                            for p in (perms.get("results") or [])
                        )
                        if is_anon:
                            out.append(AtlassianPublicSpaceRow(
                                site_url=site_url,
                                space_key=str(key),
                                space_name=sp.get("name"),
                                space_type="confluence_space",
                                public_url=(
                                    f"{site_base}/wiki/spaces/{key}"
                                ),
                                raw={"space": sp, "permissions": perms},
                            ))
                    if body.get("size", 0) < 100:
                        break
                    start += 100
            except httpx.HTTPStatusError as exc:
                if exc.response.status_code in (403, 404):
                    pass  # Confluence absent / no permission
                else:
                    raise

            # ── Jira projects with anonymous role members ──────
            try:
                start = 0
                while True:
                    resp = await client.get(
                        f"{site_base}/rest/api/3/project/search"
                        f"?startAt={start}&maxResults=50",
                        headers=headers,
                    )
                    resp.raise_for_status()
                    body = resp.json() or {}
                    for proj in body.get("values") or []:
                        proj_key = proj.get("key")
                        if not proj_key:
                            continue
                        # Browse project permission scheme — anonymous
                        # access shows up in /project/{key}/role with
                        # ``Anonymous`` role.
                        roles_resp = await client.get(
                            f"{site_base}/rest/api/3/project/"
                            f"{proj_key}/role",
                            headers=headers,
                        )
                        if roles_resp.status_code != 200:
                            continue
                        roles_map = roles_resp.json() or {}
                        anon_url = next(
                            (
                                v for k, v in roles_map.items()
                                if "anonymous" in k.lower()
                            ),
                            None,
                        )
                        if not anon_url:
                            continue
                        actors_resp = await client.get(
                            anon_url, headers=headers,
                        )
                        if actors_resp.status_code != 200:
                            continue
                        actors_payload = actors_resp.json() or {}
                        if actors_payload.get("actors"):
                            out.append(AtlassianPublicSpaceRow(
                                site_url=site_url,
                                space_key=str(proj_key),
                                space_name=proj.get("name"),
                                space_type="jira_project",
                                public_url=(
                                    f"{site_base}/jira/projects/{proj_key}"
                                ),
                                raw={"project": proj},
                            ))
                    if body.get("isLast", True):
                        break
                    start += body.get("maxResults", 50)
            except httpx.HTTPStatusError as exc:
                if exc.response.status_code in (403, 404):
                    pass
                else:
                    raise
        return out

    @classmethod
    async def list_organization_api_tokens(
        cls, session: AsyncSession,
    ) -> list[AtlassianApiTokenRow]:
        """Org-wide API token inventory via /admin/v1/orgs/{orgId}/api-tokens.

        Returns one row per token across all org users — the auditor's
        PAT module dedupes / risk-flags from there. Atlassian Guard /
        Atlassian Access is required for this endpoint; without it the
        call returns 403 and we raise IntegrationMissingError so the
        auditor honestly surfaces the gap.
        """
        import httpx

        org_id, bearer = await cls._resolve_org_admin_creds(session)
        headers = cls._admin_headers(bearer)

        async with httpx.AsyncClient(timeout=30.0, trust_env=False) as client:
            try:
                rows = await cls._paginate_admin(
                    client,
                    f"{_ATLASSIAN_ADMIN_BASE}/orgs/{org_id}/api-tokens",
                    headers,
                )
            except httpx.HTTPStatusError as exc:
                if exc.response.status_code in (403, 404):
                    raise IntegrationMissingError(
                        f"{cls.KIND} ({cls.PROVIDER}: api-tokens "
                        f"endpoint requires Atlassian Guard / Access)"
                    ) from exc
                raise

        out: list[AtlassianApiTokenRow] = []
        for r in rows:
            attrs = r.get("attributes") or r
            user = (r.get("relationships") or {}).get("user") or {}
            account_id = (
                (user.get("data") or {}).get("id")
                or attrs.get("account_id")
                or ""
            )
            out.append(AtlassianApiTokenRow(
                token_id=str(r.get("id") or attrs.get("id") or ""),
                user_account_id=str(account_id),
                user_email=(attrs.get("email") or "").lower() or None,
                label=attrs.get("label"),
                created_at=_parse_iso(attrs.get("created_at")),
                last_used_at=_parse_iso(attrs.get("last_used_at")),
                raw=r,
            ))
        return out


# ═════════════════════════════════════════════════════════════════════
# Aliyun (re-uses ``cloud/aliyun`` integration credentials)
# ═════════════════════════════════════════════════════════════════════


@dataclass(slots=True)
class AliyunAccessKeyRow:
    user_name: str
    access_key_id: str
    create_date: datetime | None
    last_used_at: datetime | None
    status: str  # "Active" | "Inactive"
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class AliyunDomainRow:
    domain_name: str
    expiration_date: datetime | None
    registration_date: datetime | None
    icp_filing_status: str | None
    icp_filing_no: str | None
    raw: dict[str, Any] = field(default_factory=dict)


class AliyunAuditorAdapter:
    """Auditor-side reads from the existing ``cloud/aliyun`` integration.

    Re-uses the same access_key_id + access_key_secret the cloud sync
    already encrypts. No extra credentials, no extra integration kind.
    """

    KIND = CLOUD_KIND
    PROVIDER = "aliyun"

    @classmethod
    async def _creds(
        cls, session: AsyncSession,
    ) -> tuple[uuid.UUID, str, str]:
        """``(integration_id, access_key_id, access_key_secret)``.

        The id travels with the credentials because the cloud drivers
        take it as a required keyword — constructing one positionally
        (as this file used to) raises TypeError before any API call.
        """
        row, config = await _resolve_integration(
            session, kind=cls.KIND, provider=cls.PROVIDER,
        )
        ak = (config.get("access_key_id") or "").strip()
        sk = (config.get("access_key_secret") or "").strip()
        if not ak or not sk:
            raise IntegrationMissingError(cls.KIND, cls.PROVIDER)
        return row.id, ak, sk

    @classmethod
    async def list_ram_access_keys(
        cls, session: AsyncSession,
    ) -> list[AliyunAccessKeyRow]:
        """Walk RAM users → list each user's AccessKeys via the
        AliyunCloudDriver's ``_acs3_call`` helper (the canonical ACS3
        v3 signing path; we don't reimplement signing).
        """
        from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver

        integration_id, ak, sk = await cls._creds(session)
        driver = AliyunCloudDriver(
            integration_id=str(integration_id),
            config={"access_key_id": ak, "access_key_secret": sk},
        )
        rows: list[AliyunAccessKeyRow] = []
        try:
            # ListUsers is Marker-paginated. Without the loop the ledger
            # stopped at the first 200 RAM users and every key past that
            # point silently escaped rotation governance — a blind spot
            # that grows with headcount. Same shape as the domain /
            # Tencent / Azure walkers below.
            users: list[dict[str, Any]] = []
            marker: str | None = None
            for _page in range(_MAX_PAGES):
                params = {"MaxItems": "200"}
                if marker:
                    params["Marker"] = marker
                users_resp = await driver._acs3_call(
                    host="ram.aliyuncs.com",
                    version="2015-05-01",
                    action="ListUsers",
                    method="POST",
                    params=params,
                ) or {}
                page_users = (
                    users_resp.get("Users", {}).get("User", [])
                    if isinstance(users_resp.get("Users"), dict)
                    else users_resp.get("Users") or []
                )
                users.extend(page_users)
                next_marker = (users_resp.get("Marker") or "").strip()
                if not users_resp.get("IsTruncated") or not next_marker:
                    break
                if next_marker == marker:
                    # Non-advancing cursor — bail rather than spin.
                    logger.warning(
                        "Aliyun ListUsers marker did not advance (%s); "
                        "stopping pagination", next_marker,
                    )
                    break
                marker = next_marker
            else:
                logger.warning(
                    "Aliyun ListUsers hit the %d-page cap — RAM user "
                    "ledger may be truncated", _MAX_PAGES,
                )
            for u in users:
                user_name = u.get("UserName")
                if not user_name:
                    continue
                ak_resp = await driver._acs3_call(
                    host="ram.aliyuncs.com",
                    version="2015-05-01",
                    action="ListAccessKeys",
                    method="POST",
                    params={"UserName": user_name},
                )
                ak_list = (
                    (ak_resp or {}).get("AccessKeys", {}).get(
                        "AccessKey", []
                    )
                    if isinstance(
                        (ak_resp or {}).get("AccessKeys"), dict
                    )
                    else (ak_resp or {}).get("AccessKeys") or []
                )
                for k in ak_list:
                    rows.append(AliyunAccessKeyRow(
                        user_name=user_name,
                        access_key_id=k.get("AccessKeyId") or "",
                        create_date=_parse_iso(k.get("CreateDate")),
                        last_used_at=_parse_iso(
                            k.get("LastUsedDate")
                        ),
                        status=k.get("Status") or "Unknown",
                        raw={"user": u, "key": k},
                    ))
        finally:
            await driver.close()
        return rows

    @classmethod
    async def list_domains(
        cls, session: AsyncSession,
    ) -> list[AliyunDomainRow]:
        """List Aliyun-registered domains with expiration + ICP fields,
        directly via the Domain service ``QueryDomainList`` API. The
        cloud driver doesn't expose a typed domain method (cmdb sync
        uses a different code path), so we drive the call ourselves
        via the driver's signing helper.
        """
        from app.integrations.drivers.aliyun_cloud import AliyunCloudDriver

        integration_id, ak, sk = await cls._creds(session)
        driver = AliyunCloudDriver(
            integration_id=str(integration_id),
            config={"access_key_id": ak, "access_key_secret": sk},
        )
        rows: list[AliyunDomainRow] = []
        try:
            page_num = 1
            while True:
                resp = await driver._acs3_call(
                    host="domain.aliyuncs.com",
                    version="2018-01-29",
                    action="QueryDomainList",
                    method="POST",
                    params={
                        "PageNum": str(page_num),
                        "PageSize": "100",
                    },
                )
                data = (resp or {}).get("Data") or {}
                domain_list = data.get("Domain") or []
                for d in domain_list:
                    rows.append(AliyunDomainRow(
                        domain_name=d.get("DomainName") or "",
                        expiration_date=_parse_iso(
                            d.get("ExpirationDate")
                            or d.get("ExpirationCurrDate")
                        ),
                        registration_date=_parse_iso(
                            d.get("RegistrationDate")
                        ),
                        icp_filing_status=d.get(
                            "DomainNameVerificationStatus"
                        ),
                        icp_filing_no=d.get("ICPFilingNumber"),
                        raw=d,
                    ))
                total_count = int(resp.get("TotalItemNum") or 0)
                if page_num * 100 >= total_count:
                    break
                page_num += 1
        finally:
            await driver.close()
        return rows


# ═════════════════════════════════════════════════════════════════════
# Tencent (re-uses ``cloud/tencent`` integration credentials)
# ═════════════════════════════════════════════════════════════════════


@dataclass(slots=True)
class TencentAccessKeyRow:
    user_name: str
    secret_id: str
    create_time: datetime | None
    last_used_at: datetime | None
    status: str
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class TencentDomainRow:
    domain_name: str
    expiration_date: datetime | None
    registration_date: datetime | None
    raw: dict[str, Any] = field(default_factory=dict)


class TencentAuditorAdapter:
    """Auditor-side reads from the existing ``cloud/tencent`` integration."""

    KIND = CLOUD_KIND
    PROVIDER = "tencent"

    @classmethod
    async def _creds(
        cls, session: AsyncSession,
    ) -> tuple[uuid.UUID, str, str]:
        """``(integration_id, secret_id, secret_key)`` — see the Aliyun
        twin for why the id comes along."""
        row, config = await _resolve_integration(
            session, kind=cls.KIND, provider=cls.PROVIDER,
        )
        sid = (config.get("secret_id") or "").strip()
        skey = (config.get("secret_key") or "").strip()
        if not sid or not skey:
            raise IntegrationMissingError(cls.KIND, cls.PROVIDER)
        return row.id, sid, skey

    @classmethod
    async def list_cam_access_keys(
        cls, session: AsyncSession,
    ) -> list[TencentAccessKeyRow]:
        """List CAM users + their access keys via the official
        ``tencentcloud-sdk-python`` CAM client. The Tencent SDK is
        sync-only, so each call runs through ``asyncio.to_thread``.
        """
        from tencentcloud.cam.v20190116 import models as cam_models  # type: ignore[import-untyped]

        from app.integrations.drivers.tencent_cloud import TencentCloudDriver

        integration_id, sid, skey = await cls._creds(session)
        driver = TencentCloudDriver(
            integration_id=str(integration_id),
            config={"secret_id": sid, "secret_key": skey},
        )
        rows: list[TencentAccessKeyRow] = []
        try:
            cam = driver._cam_client()

            def _list_users() -> list[dict[str, Any]]:
                req = cam_models.ListUsersRequest()
                resp = cam.ListUsers(req)
                # SDK objects expose ``_serialize`` via ``to_json_string``;
                # use the public ``__dict__`` shape filtered for camelCase
                # response fields.
                payload = json.loads(resp.to_json_string())
                return payload.get("Data") or []

            users = await asyncio.to_thread(_list_users)
            for u in users:
                uin = u.get("Uin")
                uname = u.get("Name")
                if not uname or uin is None:
                    continue

                def _list_keys(target_uin: int) -> list[dict[str, Any]]:
                    req = cam_models.ListAccessKeysRequest()
                    req.TargetUin = int(target_uin)
                    resp = cam.ListAccessKeys(req)
                    payload = json.loads(resp.to_json_string())
                    return payload.get("AccessKeys") or []

                ak_list = await asyncio.to_thread(_list_keys, uin)
                for k in ak_list:
                    rows.append(TencentAccessKeyRow(
                        user_name=uname,
                        secret_id=k.get("AccessKeyId") or "",
                        create_time=_parse_iso(k.get("CreateTime")),
                        last_used_at=_parse_iso(
                            k.get("LastUsedDate")
                            or k.get("LastUsed")
                        ),
                        status=k.get("Status") or "Unknown",
                        raw={"user": u, "key": k},
                    ))
        finally:
            await driver.close()
        return rows

    @classmethod
    async def list_domains(
        cls, session: AsyncSession,
    ) -> list[TencentDomainRow]:
        """List Tencent-registered domains with expiration via the
        ``domain.tencentcloudapi.com`` ``DescribeDomainNameList`` API.
        """
        from tencentcloud.common import credential
        from tencentcloud.common.profile.client_profile import (
            ClientProfile,  # type: ignore[import-untyped]
        )
        from tencentcloud.common.profile.http_profile import (
            HttpProfile,  # type: ignore[import-untyped]
        )
        from tencentcloud.domain.v20180808 import (  # type: ignore[import-untyped]
            domain_client,
        )
        from tencentcloud.domain.v20180808 import (
            models as domain_models,
        )

        _integration_id, sid, skey = await cls._creds(session)
        rows: list[TencentDomainRow] = []

        def _list_all() -> list[dict[str, Any]]:
            cred = credential.Credential(sid, skey)
            http_p = HttpProfile()
            http_p.endpoint = "domain.tencentcloudapi.com"
            http_p.reqTimeout = 30
            client_p = ClientProfile()
            client_p.httpProfile = http_p
            client = domain_client.DomainClient(cred, "", client_p)
            collected: list[dict[str, Any]] = []
            offset = 0
            page = 100
            while True:
                req = domain_models.DescribeDomainNameListRequest()
                req.Offset = offset
                req.Limit = page
                resp = client.DescribeDomainNameList(req)
                payload = json.loads(resp.to_json_string())
                items = payload.get("DomainSet") or []
                collected.extend(items)
                total = int(payload.get("TotalCount") or 0)
                offset += page
                if offset >= total or not items:
                    break
            return collected

        domains = await asyncio.to_thread(_list_all)
        for d in domains:
            rows.append(TencentDomainRow(
                domain_name=d.get("DomainName") or "",
                expiration_date=_parse_iso(
                    d.get("ExpirationDate")
                ),
                registration_date=_parse_iso(
                    d.get("CreationDate")
                ),
                raw=d,
            ))
        return rows


# ═════════════════════════════════════════════════════════════════════
# Azure (re-uses ``cloud/azure`` integration credentials)
# ═════════════════════════════════════════════════════════════════════


@dataclass(slots=True)
class AzureStorageKeyRow:
    subscription_id: str
    resource_group: str
    storage_account_name: str
    key_name: str           # "key1" | "key2"
    last_rotated_at: datetime | None
    raw: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class AzureAppCredentialRow:
    application_id: str
    application_display_name: str | None
    credential_id: str
    credential_display_name: str | None
    start_date: datetime | None
    end_date: datetime | None
    raw: dict[str, Any] = field(default_factory=dict)


class AzureAuditorAdapter:
    """Auditor-side reads from the existing ``cloud/azure`` integration."""

    KIND = CLOUD_KIND
    PROVIDER = "azure"

    @classmethod
    async def _creds(
        cls, session: AsyncSession,
    ) -> tuple[uuid.UUID, str, str, str]:
        """``(integration_id, tenant_id, client_id, client_secret)`` —
        see the Aliyun twin for why the id comes along."""
        row, config = await _resolve_integration(
            session, kind=cls.KIND, provider=cls.PROVIDER,
        )
        tenant_id = (config.get("tenant_id") or "").strip()
        client_id = (config.get("client_id") or "").strip()
        client_secret = (config.get("client_secret") or "").strip()
        if not (tenant_id and client_id and client_secret):
            raise IntegrationMissingError(cls.KIND, cls.PROVIDER)
        return row.id, tenant_id, client_id, client_secret

    @classmethod
    async def list_storage_account_keys(
        cls, session: AsyncSession,
    ) -> list[AzureStorageKeyRow]:
        """Walk every Storage Account in every accessible subscription
        and record (key1, key2) creation timestamps.

        Uses the AzureCloudDriver's ``_http()`` httpx client and
        ``_token(resource)`` for AAD client_credentials auth. ARM
        pagination is ``@odata.nextLink`` walk.
        """
        import httpx as _httpx

        from app.integrations.drivers.azure_cloud import AzureCloudDriver

        integration_id, tid, cid, csecret = await cls._creds(session)
        driver = AzureCloudDriver(
            integration_id=str(integration_id),
            config={
                "tenant_id": tid,
                "client_id": cid,
                "client_secret": csecret,
            },
        )
        rows: list[AzureStorageKeyRow] = []
        try:
            arm_token = await driver._token(
                "https://management.azure.com",
            )
            arm_headers = {
                "Authorization": f"Bearer {arm_token}",
                "Accept": "application/json",
            }
            subs = await driver.list_scopes()
            async with _httpx.AsyncClient(timeout=30.0, trust_env=False) as http:
                for sub in subs:
                    sub_id = sub.external_id
                    # List all storage accounts in the subscription —
                    # ARM cross-resource-group endpoint with paging.
                    next_url: str | None = (
                        f"https://management.azure.com/subscriptions/"
                        f"{sub_id}/providers/Microsoft.Storage/"
                        f"storageAccounts?api-version=2023-05-01"
                    )
                    while next_url:
                        resp = await http.get(next_url, headers=arm_headers)
                        if resp.status_code in (401, 403):
                            # Subscription not accessible to this SP —
                            # skip; auditor honestly returns rows only
                            # for subs we can read.
                            break
                        resp.raise_for_status()
                        body = resp.json() or {}
                        for acc in body.get("value") or []:
                            name = acc.get("name") or ""
                            arm_id = acc.get("id") or ""
                            rg = (
                                arm_id.split("/resourceGroups/")[-1]
                                .split("/")[0]
                                if "/resourceGroups/" in arm_id else ""
                            )
                            if not (name and rg):
                                continue
                            # listKeys is POST and returns key metadata
                            # including ``creationTime`` per key. The
                            # ``value`` field (actual secret) is in the
                            # response — we deliberately strip it from
                            # raw so we don't persist key material.
                            keys_resp = await http.post(
                                f"https://management.azure.com/"
                                f"subscriptions/{sub_id}/"
                                f"resourceGroups/{rg}/providers/"
                                f"Microsoft.Storage/storageAccounts/"
                                f"{name}/listKeys"
                                f"?api-version=2023-05-01",
                                headers=arm_headers,
                                content=b"",
                            )
                            if keys_resp.status_code in (401, 403, 404):
                                continue
                            keys_resp.raise_for_status()
                            keys_body = keys_resp.json() or {}
                            for k in keys_body.get("keys") or []:
                                rows.append(AzureStorageKeyRow(
                                    subscription_id=sub_id,
                                    resource_group=rg,
                                    storage_account_name=name,
                                    key_name=k.get("keyName") or "",
                                    last_rotated_at=_parse_iso(
                                        k.get("creationTime")
                                    ),
                                    raw={
                                        "account_id": arm_id,
                                        # Strip ``value`` — never persist
                                        # secret material; the auditor
                                        # cares about rotation timing.
                                        "key_meta": {
                                            kk: vv
                                            for kk, vv in k.items()
                                            if kk != "value"
                                        },
                                    },
                                ))
                        next_url = body.get("nextLink")
        finally:
            await driver.close()
        return rows

    @classmethod
    async def list_app_password_credentials(
        cls, session: AsyncSession,
    ) -> list[AzureAppCredentialRow]:
        """Service-principal client secrets via Microsoft Graph
        ``/applications``. One row per secret per app; ``end_date`` is
        what the auditor compares for rotation / expiry alerting.

        The Graph App requires ``Application.Read.All`` permission. If
        absent (401/403), surfaces honestly via IntegrationMissingError
        so the auditor reports the missing scope instead of empty data.
        """
        import httpx as _httpx

        from app.integrations.drivers.azure_cloud import AzureCloudDriver

        integration_id, tid, cid, csecret = await cls._creds(session)
        driver = AzureCloudDriver(
            integration_id=str(integration_id),
            config={
                "tenant_id": tid,
                "client_id": cid,
                "client_secret": csecret,
            },
        )
        rows: list[AzureAppCredentialRow] = []
        try:
            graph_token = await driver._token(
                "https://graph.microsoft.com",
            )
            headers = {
                "Authorization": f"Bearer {graph_token}",
                "Accept": "application/json",
            }
            next_url: str | None = (
                "https://graph.microsoft.com/v1.0/applications"
                "?$select=id,appId,displayName,passwordCredentials"
                "&$top=200"
            )
            async with _httpx.AsyncClient(timeout=30.0, trust_env=False) as http:
                while next_url:
                    resp = await http.get(next_url, headers=headers)
                    if resp.status_code in (401, 403):
                        raise IntegrationMissingError(
                            f"{cls.KIND} ({cls.PROVIDER}: Graph "
                            f"Application.Read.All scope required for "
                            f"SP credential enumeration)"
                        )
                    resp.raise_for_status()
                    body = resp.json() or {}
                    for app in body.get("value") or []:
                        app_id = app.get("id") or app.get("appId") or ""
                        display = app.get("displayName")
                        for cred in app.get("passwordCredentials") or []:
                            rows.append(AzureAppCredentialRow(
                                application_id=str(app_id),
                                application_display_name=display,
                                credential_id=str(cred.get("keyId") or ""),
                                credential_display_name=cred.get(
                                    "displayName"
                                ),
                                start_date=_parse_iso(
                                    cred.get("startDateTime")
                                ),
                                end_date=_parse_iso(
                                    cred.get("endDateTime")
                                ),
                                raw=cred,
                            ))
                    next_url = body.get("@odata.nextLink")
        finally:
            await driver.close()
        return rows


# ═════════════════════════════════════════════════════════════════════
# Feishu signin event extension
# (re-uses the existing ``chat_approval/feishu`` integration)
# ═════════════════════════════════════════════════════════════════════


@dataclass(slots=True)
class FeishuSigninRow:
    event_id: str
    user_email: str | None
    open_id: str | None
    event_at: datetime
    event_kind: str
    status: str
    source_ip: str | None
    user_agent: str | None
    raw: dict[str, Any]


# Feishu audit log event types we treat as login events.
# Reference: open.feishu.cn → 通讯录 → 审计日志 → event_name catalog.
_FEISHU_LOGIN_EVENT_NAMES = frozenset({
    "user_login",          # standard login
    "user_login_failed",   # failed login
    "user_2fa_request",    # 2FA prompt
    "admin_login",         # admin console login
})


async def _feishu_audit_list(
    cls,
    session: AsyncSession,
    *,
    start_at: datetime,
    end_at: datetime,
) -> list[FeishuSigninRow]:
    """Pull login-like events from Feishu's audit log endpoint
    ``/open-apis/admin/v1/audit_infos``. Requires the Feishu app to
    have the ``admin:audit:readonly`` permission and to be attached to
    the tenant's admin scope.
    """
    import httpx

    token, _config = await FeishuAuditorAdapter._tenant_token(session)

    rows: list[FeishuSigninRow] = []
    page_token: str | None = None
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json; charset=utf-8",
    }
    base = "https://open.feishu.cn/open-apis/admin/v1/audit_infos"
    async with httpx.AsyncClient(timeout=30.0, trust_env=False) as client:
        while True:
            params: dict[str, Any] = {
                "latest": int(end_at.timestamp()),
                "oldest": int(start_at.timestamp()),
                "page_size": 200,
            }
            if page_token:
                params["page_token"] = page_token
            resp = await client.get(base, headers=headers, params=params)
            if resp.status_code in (401, 403):
                raise IntegrationMissingError(
                    "chat_approval (feishu: audit_infos requires "
                    "admin:audit:readonly scope)"
                )
            resp.raise_for_status()
            body = (resp.json() or {}).get("data") or {}
            for item in body.get("items") or []:
                event_name = item.get("event_name") or ""
                if event_name not in _FEISHU_LOGIN_EVENT_NAMES:
                    continue
                operator = item.get("operator_value") or {}
                ts = item.get("event_time")
                if isinstance(ts, (int, float)):
                    event_at = datetime.fromtimestamp(
                        float(ts), tz=UTC,
                    )
                else:
                    parsed = _parse_iso(str(ts) if ts else None)
                    event_at = parsed or datetime.now(UTC)
                client_info = item.get("event_module") or {}
                rows.append(FeishuSigninRow(
                    event_id=str(item.get("audit_log_id")
                                  or item.get("event_id")
                                  or f"feishu:{ts}:{operator}"),
                    user_email=_feishu_email(operator),
                    open_id=operator.get("open_id"),
                    event_at=event_at,
                    event_kind=event_name,
                    status=(
                        "failure" if "failed" in event_name
                        else "success"
                    ),
                    source_ip=item.get("ip"),
                    user_agent=str(client_info) if client_info else None,
                    raw=item,
                ))
            if not body.get("has_more"):
                break
            page_token = body.get("page_token")
            if not page_token:
                break
    return rows


# Attach as a classmethod onto the existing FeishuAuditorAdapter so
# the sync code can call FeishuAuditorAdapter.list_signin_events(...)
# uniformly with the Google adapter.
FeishuAuditorAdapter.list_signin_events = classmethod(  # type: ignore[attr-defined]
    _feishu_audit_list
)