"""GitHub (Enterprise Cloud + SAML) software-provisioning driver.

Uses GitHub's SCIM 2.0 API for organization user provisioning. Works
on GitHub Enterprise Cloud with SAML SSO enabled at the org level;
for non-Enterprise public orgs, a different pattern
(admin-maintained email→handle mapping) would be required and is
out of scope for this PR.

Per-binding configuration is minimal — GitHub doesn't have user
types or tiers the way Zoom does. Everything is scoped to the
organization specified in the integration config.

API reference:
  https://docs.github.com/en/enterprise-cloud@latest/rest/scim
"""
from __future__ import annotations

import asyncio
import logging
import re
from typing import Any
from urllib.parse import quote as _url_quote

import httpx

from app.integrations.driver_base import (
    ProvisioningContext,
    ProvisioningResult,
    SoftwareProvisioningDriver,
)
from app.integrations.registry import register_driver
from app.integrations.schemas import (
    IntegrationFieldInfo,
    IntegrationTestResult,
)

logger = logging.getLogger(__name__)

_DEFAULT_BASE = "https://api.github.com"
_HTTP_TIMEOUT = httpx.Timeout(30.0, connect=10.0)
_SCIM_CONTENT = "application/scim+json"


def _escape_scim_filter(value: str) -> str:
    """Escape a string for inclusion inside a SCIM filter literal.

    SCIM (RFC 7644 §3.4.2) uses ASCII double-quotes around literals
    and supports backslash-escaped quotes. A poisoned input like
    ``foo"or 1 eq 1"@bar.com`` would otherwise break out of the
    quoted literal and turn the filter into a tautology. The
    work_email column is HRIS-controlled in normal operation, but
    multi-vendor sync paths feed it externally — defence-in-depth
    matches what slack_software.py already does for the same surface.
    """
    return value.replace("\\", "\\\\").replace('"', '\\"')


def _redact_secrets(blob: str, *known_secrets: str) -> str:
    """Strip Authorization tokens / API keys from error-snippet blobs
    before they're persisted into ``provisioning_tasks.error_message``
    (visible to admin & asset_admin) or
    ``software_connectors.last_error_message``."""
    if not blob:
        return blob
    for secret in known_secrets:
        if secret:
            blob = blob.replace(secret, "***")
    blob = re.sub(
        r"(?i)(authorization\s*:?\s*(?:bearer|basic)\s+)[A-Za-z0-9._\-+/=]+",
        r"\1***",
        blob,
    )
    blob = re.sub(r"\bghp_[A-Za-z0-9]{20,}\b", "ghp_***", blob)
    blob = re.sub(r"\bghs_[A-Za-z0-9]{20,}\b", "ghs_***", blob)
    blob = re.sub(r"\bghu_[A-Za-z0-9]{20,}\b", "ghu_***", blob)
    blob = re.sub(r"\bgithub_pat_[A-Za-z0-9_]{20,}\b", "github_pat_***", blob)
    return blob


@register_driver
class GitHubSoftwareDriver(SoftwareProvisioningDriver):
    provider = "github"
    display_name = "GitHub (Organization)"
    description = (
        "GitHub Enterprise Cloud with SAML SSO — 通过 SCIM 2.0 管理"
        "组织成员。需要 admin:org + scim:enterprise 权限的 token。"
        "grant = 邀请/激活组织成员; revoke = deprovision (停用)。"
        "本轮只做组织级,不处理 team / repo 权限。"
    )
    supports_seat = True
    supports_permission = False
    supports_multi_instance = True
    config_schema = [
        IntegrationFieldInfo(
            key="org_slug", label="Organization Slug",
            is_secret=False, required=True,
            placeholder="acme-inc",
            hint="GitHub org URL 最后那段(github.com/<slug>)。",
        ),
        IntegrationFieldInfo(
            key="token", label="API Token",
            is_secret=True, required=True,
            placeholder="ghp_... / github_pat_...",
            hint="Personal access token 或 GitHub App token,"
                 "需要 admin:org + scim:enterprise 权限。",
        ),
        IntegrationFieldInfo(
            key="base_url", label="API Base URL",
            is_secret=False, required=False,
            default=_DEFAULT_BASE,
            placeholder=_DEFAULT_BASE,
            hint="通常留空;GitHub Enterprise Server 自托管才需要改。",
        ),
    ]
    binding_schema = []  # org-level; no per-software params

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        self._safe_target: Any | None = None
        self._safe_target_url: str | None = None
        self._safe_target_lock = asyncio.Lock()

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

    def _base_url(self) -> str:
        url = str(self.config.get("base_url") or _DEFAULT_BASE).rstrip("/")
        return url

    async def _transport(self):
        from app.integrations._url_safety import (
            PinnedAsyncHTTPTransport,
            resolve_safe_external_url_async,
        )
        url = self._base_url()
        if self._safe_target_url != url:
            async with self._safe_target_lock:
                if self._safe_target_url != url:
                    target = await resolve_safe_external_url_async(url)
                    self._safe_target = target
                    self._safe_target_url = url
        # The cache is populated before its URL marker, so this can only be
        # None if future maintenance violates that ordering.
        if self._safe_target is None:
            raise RuntimeError("GitHub safe URL target was not initialized")
        return PinnedAsyncHTTPTransport(self._safe_target)

    def _org(self) -> str:
        v = self.config.get("org_slug")
        if not v:
            raise ValueError("GitHub integration missing org_slug")
        return str(v)

    def _token(self) -> str:
        v = self.config.get("token")
        if not v:
            raise ValueError("GitHub integration missing token")
        return str(v)

    def _headers(self) -> dict[str, str]:
        return {
            "Authorization": f"Bearer {self._token()}",
            "Accept": _SCIM_CONTENT,
            "Content-Type": _SCIM_CONTENT,
            "X-GitHub-Api-Version": "2022-11-28",
        }

    def _scim_users_url(self) -> str:
        return f"{self._base_url()}/scim/v2/organizations/{_url_quote(self._org(), safe='')}/Users"

    def _build_payload(self, ctx: ProvisioningContext) -> dict[str, Any]:
        display = (ctx.user_display_name or ctx.user_email.split("@", 1)[0]).strip()
        first, _, last = display.partition(" ")
        return {
            "schemas": ["urn:ietf:params:scim:schemas:core:2.0:User"],
            "userName": ctx.user_email,
            "name": {
                "givenName": first or ctx.user_email,
                "familyName": last or "-",
            },
            "emails": [{"value": ctx.user_email, "type": "work", "primary": True}],
            "active": True,
        }

    async def _find_user_id(
        self, email: str,
    ) -> tuple[str | None, httpx.Response | None, str | None]:
        """Look up a SCIM user id by email.

        Returns ``(id, raw_response, network_error)``:

        * ``id`` is ``None`` when the SCIM filter returned 0 results.
        * ``raw_response`` is the underlying ``httpx.Response`` when
          the call reached GitHub; ``None`` only when an httpx-level
          transport failure occurred (timeout, DNS, TLS, connection
          reset).
        * ``network_error`` is a short error label (``f"{type}: {exc}"``)
          when the lookup itself failed at the transport layer; callers
          MUST NOT interpret this case as "user is already absent" —
          surface it as a transient sync error so the next tick retries.

        Pre-launch P1 fix (round-19 audit, 2026-05-10): the prior
        signature collapsed httpx-transport failures and "user not
        found" into the same ``(None, None)`` tuple. ``revoke_seat``
        then treated network errors as ``already_absent=True`` and
        silently confirmed the revoke against a still-active GitHub
        org member — the local seat row flipped to ``revoked`` while
        the real org membership lived on (billing leak + access leak).
        """
        try:
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT,
                trust_env=False,
                transport=await self._transport(),
            ) as client:
                resp = await client.get(
                    self._scim_users_url(),
                    headers=self._headers(),
                    params={
                        "filter": f'userName eq "{_escape_scim_filter(email)}"',
                    },
                )
        except httpx.HTTPError as exc:
            return None, None, _redact_secrets(
                f"{type(exc).__name__}: {exc}", self._token(),
            )
        if resp.status_code != 200:
            return None, resp, None
        data = resp.json() or {}
        resources = data.get("Resources") or []
        if not resources:
            return None, resp, None
        return str(resources[0].get("id")), resp, None

    # ── test_connection ──────────────────────────────────────────

    async def test_connection(self) -> IntegrationTestResult:
        """Probe: list up to 1 user in the org's SCIM namespace."""
        try:
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT,
                trust_env=False,
                transport=await self._transport(),
            ) as client:
                resp = await client.get(
                    self._scim_users_url(),
                    headers=self._headers(),
                    params={"count": 1},
                )
        except httpx.HTTPError as e:
            return IntegrationTestResult(
                ok=False,
                message=_redact_secrets(f"{type(e).__name__}: {e}", self._token()),
            )
        except Exception as e:  # noqa: BLE001
            return IntegrationTestResult(
                ok=False,
                message=_redact_secrets(f"{type(e).__name__}: {e}", self._token()),
            )
        if resp.status_code == 404:
            return IntegrationTestResult(
                ok=False,
                message=(
                    "GitHub returned 404 — SCIM requires GitHub "
                    "Enterprise Cloud with SAML SSO enabled at the org. "
                    "Check org slug and token scopes."
                ),
                detail={"status": 404},
            )
        if resp.status_code != 200:
            return IntegrationTestResult(
                ok=False,
                message=(
                    f"GitHub SCIM {resp.status_code}: "
                    f"{_redact_secrets(resp.text[:300], self._token())}"
                ),
                detail={"status": resp.status_code},
            )
        data = resp.json() or {}
        total = int(data.get("totalResults") or 0)
        return IntegrationTestResult(
            ok=True,
            message=f"OK · {total} members in {self._org()}",
            detail={"totalResults": total, "org": self._org()},
        )

    # ── grant_seat ───────────────────────────────────────────────

    async def grant_seat(self, ctx: ProvisioningContext) -> ProvisioningResult:
        """Provision a user into the GitHub org via SCIM.

        GitHub returns 201 for new SCIM records and 409 if the user
        already exists. On 409 we look them up and, if deprovisioned
        (active=false), PATCH them back to active=true.
        """
        payload = self._build_payload(ctx)
        try:
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT,
                trust_env=False,
                transport=await self._transport(),
            ) as client:
                resp = await client.post(
                    self._scim_users_url(),
                    headers=self._headers(),
                    json=payload,
                )
        except httpx.HTTPError as e:
            return ProvisioningResult(
                ok=False,
                error_message=_redact_secrets(
                    f"{type(e).__name__}: {e}", self._token(),
                ),
            )

        if resp.status_code in (200, 201):
            data = resp.json() or {}
            return ProvisioningResult(
                ok=True,
                external_ref=str(data.get("id") or ctx.user_email),
                raw_response=data,
            )
        if resp.status_code == 409:
            return await self._reactivate(ctx)

        body_preview = _redact_secrets(resp.text[:300], self._token())
        return ProvisioningResult(
            ok=False,
            error_message=(
                f"GitHub SCIM provision {resp.status_code}: {body_preview}"
            ),
            raw_response={"status": resp.status_code, "body": body_preview},
        )

    async def _reactivate(self, ctx: ProvisioningContext) -> ProvisioningResult:
        scim_id, lookup_resp, net_err = await self._find_user_id(ctx.user_email)
        if net_err is not None:
            return ProvisioningResult(
                ok=False,
                error_message=(
                    f"GitHub SCIM lookup transport error during reactivate "
                    f"of {ctx.user_email!r}: {net_err}"
                ),
            )
        if scim_id is None:
            return ProvisioningResult(
                ok=False,
                error_message=(
                    f"GitHub returned 409 but SCIM lookup for "
                    f"{ctx.user_email!r} found 0 users "
                    f"(status={lookup_resp.status_code if lookup_resp else 'n/a'})"
                ),
            )
        # Check current active state
        if lookup_resp is not None:
            resources = (lookup_resp.json() or {}).get("Resources") or []
            if resources and bool(resources[0].get("active")):
                return ProvisioningResult(
                    ok=True,
                    external_ref=scim_id,
                    raw_response={"already_active": True},
                )

        try:
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT,
                trust_env=False,
                transport=await self._transport(),
            ) as client:
                patch = await client.patch(
                    f"{self._scim_users_url()}/{_url_quote(scim_id, safe='')}",
                    headers=self._headers(),
                    json={
                        "schemas": [
                            "urn:ietf:params:scim:api:messages:2.0:PatchOp",
                        ],
                        "Operations": [
                            {"op": "replace", "path": "active", "value": True},
                        ],
                    },
                )
        except httpx.HTTPError as e:
            return ProvisioningResult(
                ok=False,
                error_message=_redact_secrets(
                    f"GitHub reactivate failed: {type(e).__name__}: {e}",
                    self._token(),
                ),
            )
        if patch.status_code in (200, 204):
            return ProvisioningResult(
                ok=True,
                external_ref=scim_id,
                raw_response={"reactivated": True},
            )
        return ProvisioningResult(
            ok=False,
            error_message=(
                f"GitHub SCIM reactivate {patch.status_code}: "
                f"{_redact_secrets(patch.text[:200], self._token())}"
            ),
        )

    # ── revoke_seat ──────────────────────────────────────────────

    async def revoke_seat(self, ctx: ProvisioningContext) -> ProvisioningResult:
        """Deprovision: SCIM DELETE, which on GitHub suspends the
        user's org membership (SAML identity retained so a rehire can
        resume the same audit trail).

        Treats "user already absent" as success for offboarding
        idempotency.
        """
        scim_id, lookup_resp, net_err = await self._find_user_id(ctx.user_email)
        if net_err is not None:
            # Transport failure (timeout, DNS, TLS, connection reset).
            # NEVER interpret as "user already absent" — that would
            # silently confirm a revoke against a still-active GitHub
            # member. Surface as a transient error; next sync tick
            # retries.
            return ProvisioningResult(
                ok=False,
                error_message=(
                    f"GitHub SCIM lookup transport error for "
                    f"{ctx.user_email!r}: {net_err}"
                ),
            )
        if scim_id is None:
            if lookup_resp is not None and lookup_resp.status_code != 200:
                # 403/404 on the SCIM endpoint = the org exposes no usable
                # SCIM surface (not GitHub Enterprise Cloud, or SAML/SCIM
                # provisioning disabled). That's a plan/capability gate, not
                # a transient failure and not "already absent" (which comes
                # back as 200 with empty Resources). Downgrade to
                # manual_notify rather than hard-fail so a non-SCIM org
                # cannot wedge an offboarding process. Other non-200 codes
                # (5xx / 429) stay a plain failure for transient retry.
                capability_gated = lookup_resp.status_code in (403, 404)
                return ProvisioningResult(
                    ok=False,
                    capability_unsupported=capability_gated,
                    error_message=(
                        f"GitHub SCIM lookup for {ctx.user_email!r} "
                        f"returned {lookup_resp.status_code}: "
                        f"{_redact_secrets(lookup_resp.text[:200], self._token())}"
                    ),
                )
            return ProvisioningResult(
                ok=True,
                external_ref=f"github:{ctx.user_email}:already_absent",
                raw_response={"already_absent": True},
            )

        try:
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT,
                trust_env=False,
                transport=await self._transport(),
            ) as client:
                resp = await client.delete(
                    f"{self._scim_users_url()}/{_url_quote(scim_id, safe='')}",
                    headers=self._headers(),
                )
        except httpx.HTTPError as e:
            return ProvisioningResult(
                ok=False,
                error_message=_redact_secrets(
                    f"{type(e).__name__}: {e}", self._token(),
                ),
            )
        if resp.status_code in (200, 204):
            return ProvisioningResult(
                ok=True,
                external_ref=scim_id,
                raw_response={"deprovisioned": True},
            )
        if resp.status_code == 404:
            return ProvisioningResult(
                ok=True,
                external_ref=scim_id,
                raw_response={"already_absent": True},
            )
        return ProvisioningResult(
            ok=False,
            error_message=(
                f"GitHub SCIM delete {resp.status_code}: "
                f"{_redact_secrets(resp.text[:200], self._token())}"
            ),
        )