#!/usr/bin/env python3
"""Fail-closed HubSpot credential admission for an offline backend start.

The deployment database is not an environment-identity authority: a TiDB
branch reset can legitimately copy encrypted secret rows from production into
a non-production branch.  This gate verifies the *active database secret*
with HubSpot before the backend process is allowed to start.
"""
from __future__ import annotations

import argparse
import json
import os
from pathlib import Path

import deploy_offline_env
from common import (
    DeployError,
    db_exec,
    db_query_scalar,
    decrypt_named_credential_secret,
    load_env_file,
    log,
)
from deployment_plan import hubspot_credential_admission_policy, topology_target_config


HUBSPOT_CONNECTOR_OWNER_PACK = "hubspot-crm-connector"
HUBSPOT_NAMED_CREDENTIAL = "HubSpotPrivateApp"
HUBSPOT_WEBHOOK_SECRET_NAMES = ("webhook", "workflow-action")
BRANCH_RESET_ENVIRONMENT_PREFIXES = {
    "staging": "staging-",
    "test": "test-",
}
NONPROD_HUBSPOT_PUBLIC_BASE_URLS = {
    "staging": "https://nexuscrm-new-staging.pingcap.com",
    "test1": "https://nexuscrm-new-test1.pingcap.com",
    "test2": "https://nexuscrm-new-test2.pingcap.com",
    "test3": "https://nexuscrm-new-test3.pingcap.com",
}


class HubSpotCredentialQuarantineRequired(DeployError):
    """The stored secret or identity is unsafe and must be removed when bound."""


class HubSpotIdentityUnavailable(DeployError):
    """The provider could not verify identity; block without deleting a valid secret."""



def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--env-file", required=True, type=Path)
    parser.add_argument(
        "--verify-only",
        action="store_true",
        help=(
            "verify the active token identity without quarantining or mutating secrets"
        ),
    )
    parser.add_argument(
        "--target-env",
        choices=tuple(NONPROD_HUBSPOT_PUBLIC_BASE_URLS),
        help="selected public target; required with --verify-only",
    )
    return parser.parse_args()


def _scalar_count(raw: str, *, subject: str) -> int:
    try:
        return int(raw)
    except (TypeError, ValueError) as exc:
        raise HubSpotCredentialQuarantineRequired(f"invalid {subject} count") from exc


def _named_credential_filter() -> str:
    return (
        "FROM _System_NamedCredentialSecret secret "
        "JOIN _System_NamedCredential credential "
        "ON credential.Id = secret.NamedCredentialId "
        f"WHERE credential.DeveloperName = '{HUBSPOT_NAMED_CREDENTIAL}' "
        f"AND credential.OwnerPack = '{HUBSPOT_CONNECTOR_OWNER_PACK}' "
        "AND secret.SecretKind = 'Token'"
    )


def _webhook_secret_filter() -> str:
    names = ", ".join(f"'{name}'" for name in HUBSPOT_WEBHOOK_SECRET_NAMES)
    return (
        "FROM _System_WebhookSecretValue value "
        "JOIN _System_WebhookSecret secret "
        "ON secret.Id = value.WebhookSecretId "
        f"WHERE secret.OwnerPack = '{HUBSPOT_CONNECTOR_OWNER_PACK}' "
        f"AND secret.Name IN ({names}) "
        "AND COALESCE(secret.IsDeleted, 0) = 0"
    )


def nonprod_runtime_environment(envv: dict[str, str]) -> str:
    environment = str(envv.get("ENV_NAME", "")).strip().lower()
    if not environment:
        raise DeployError("HubSpot credential admission requires ENV_NAME")
    if environment in {"ci", "exp", "staging"}:
        return environment
    if any(environment.startswith(prefix) for prefix in BRANCH_RESET_ENVIRONMENT_PREFIXES.values()):
        return environment
    raise DeployError(
        "HubSpot credential admission only runs for a declared non-production runtime"
    )


def credential_policy_environment(runtime_environment: str) -> str | None:
    if runtime_environment == "exp":
        return "exp"
    if runtime_environment == "staging" or runtime_environment.startswith("staging-"):
        return "staging"
    if runtime_environment.startswith("test-"):
        return "test"
    return None


def expected_nonprod_hubspot_identity(runtime_environment: str) -> dict[str, str]:
    policy_environment = credential_policy_environment(runtime_environment)
    if policy_environment is None:
        raise DeployError(
            "an active HubSpot credential is not permitted for a non-production "
            "runtime without a committed identity policy"
        )
    policy = hubspot_credential_admission_policy(
        policy_environment,
        error_type=DeployError,
    )
    if policy["mode"] != "nonprod":
        raise DeployError(
            "non-production HubSpot credential admission requires a nonprod policy"
        )
    return {
        "portalId": policy["expected_portal_id"],
        "accountType": policy["expected_account_type"],
    }


def _database(envv: dict[str, str]) -> str:
    database = str(envv.get("TIDB_DATABASE", "")).strip()
    if not database:
        raise DeployError("HubSpot credential admission requires TIDB_DATABASE")
    return database


def _master_key(envv: dict[str, str]) -> str:
    master_key = str(envv.get("NEXUSCRM_NC_MASTER_KEY", "")).strip()
    if not master_key:
        raise HubSpotCredentialQuarantineRequired(
            "HubSpot credential admission requires NEXUSCRM_NC_MASTER_KEY"
        )
    return master_key


def _active_hubspot_access_token(
    envv: dict[str, str],
    database: str,
) -> str | None:
    credential_filter = _named_credential_filter()
    definition_count = _scalar_count(
        db_query_scalar(
            envv,
            "SELECT COUNT(*) FROM _System_NamedCredential "
            f"WHERE DeveloperName = '{HUBSPOT_NAMED_CREDENTIAL}' "
            f"AND OwnerPack = '{HUBSPOT_CONNECTOR_OWNER_PACK}';",
            database=database,
        ),
        subject="HubSpot named credential definition",
    )
    if definition_count == 0:
        return None
    if definition_count != 1:
        raise HubSpotCredentialQuarantineRequired(
            "HubSpot credential admission requires exactly one "
            "hubspot-crm-connector/HubSpotPrivateApp definition"
        )
    secret_count = _scalar_count(
        db_query_scalar(
            envv,
            f"SELECT COUNT(*) {credential_filter};",
            database=database,
        ),
        subject="HubSpot named credential secret",
    )
    if secret_count == 0:
        return None
    if secret_count != 1:
        raise HubSpotCredentialQuarantineRequired(
            "HubSpot credential admission requires exactly one active "
            "hubspot-crm-connector/HubSpotPrivateApp Token secret"
        )
    ciphertext = db_query_scalar(
        envv,
        f"SELECT secret.SecretValue {credential_filter} LIMIT 1;",
        database=database,
    )
    if not ciphertext:
        raise HubSpotCredentialQuarantineRequired(
            "HubSpot credential admission found an empty Token secret"
        )
    master_key = _master_key(envv)
    try:
        return decrypt_named_credential_secret(ciphertext, master_key)
    except DeployError as exc:
        raise HubSpotCredentialQuarantineRequired(
            "HubSpot credential admission cannot decrypt the active Token secret"
        ) from exc


def _active_hubspot_webhook_secret_count(
    envv: dict[str, str],
    database: str,
) -> int:
    return _scalar_count(
        db_query_scalar(
            envv,
            f"SELECT COUNT(*) {_webhook_secret_filter()};",
            database=database,
        ),
        subject="HubSpot webhook secret value",
    )


def hubspot_account_identity(access_token: str) -> dict[str, str]:
    try:
        payload = deploy_offline_env.hubspot_json_request(
            "/account-info/v3/details",
            access_token=access_token,
        )
    except DeployError as exc:
        raise HubSpotIdentityUnavailable(
            "HubSpot account-info identity verification request failed"
        ) from exc
    except json.JSONDecodeError as exc:
        raise HubSpotCredentialQuarantineRequired(
            "HubSpot account-info identity verification returned malformed JSON"
        ) from exc
    if not isinstance(payload, dict):
        raise HubSpotCredentialQuarantineRequired(
            "HubSpot account-info identity verification returned a non-object response"
        )
    portal_id = str(payload.get("portalId") or "").strip()
    account_type = str(payload.get("accountType") or "").strip()
    if not portal_id or not account_type:
        raise HubSpotCredentialQuarantineRequired(
            "HubSpot account-info identity verification returned an incomplete identity"
        )
    return {"portalId": portal_id, "accountType": account_type}


def verified_branch_reset_quarantine_target(envv: dict[str, str]) -> tuple[str, str]:
    """Prove the configured branch is a declared non-production target.

    The destructive quarantine is deliberately available only to branch-reset
    environments whose exact TiDB target is committed in deployment config.
    Other offline environments still fail closed on a production credential,
    but do not run a DELETE against an unbound database connection.
    """
    runtime_environment = nonprod_runtime_environment(envv)
    for environment, prefix in BRANCH_RESET_ENVIRONMENT_PREFIXES.items():
        if runtime_environment.startswith(prefix):
            target = runtime_environment.removeprefix(prefix)
            declared_target = str(envv.get("DEPLOY_TARGET", "")).strip()
            if declared_target != target:
                raise DeployError(
                    "HubSpot credential quarantine requires ENV_NAME and DEPLOY_TARGET to agree"
                )
            expected = topology_target_config(
                environment,
                target,
                error_type=DeployError,
            )
            tidb = expected.get("tidb")
            if not isinstance(tidb, dict):
                raise DeployError("HubSpot credential quarantine target has no committed TiDB identity")
            expected_values = {
                "TIDB_HOST": tidb["host"],
                "TIDB_PORT": tidb["port"],
                "TIDB_USER": tidb["user"],
                "TIDB_DATABASE": tidb["database"],
                "TIDB_CLOUD_CLUSTER_ID": tidb["cluster_id"],
                "TIDB_CLOUD_BRANCH_ID": tidb["branch_id"],
            }
            mismatches = [
                key
                for key, value in expected_values.items()
                if str(envv.get(key, "")).strip() != str(value).strip()
            ]
            if mismatches:
                raise DeployError(
                    "HubSpot credential quarantine target does not match committed "
                    f"{environment}/{target} TiDB identity: {', '.join(mismatches)}"
                )
            return environment, target
    raise DeployError(
        "HubSpot credential quarantine is only permitted for a declared TiDB branch-reset target"
    )


def scrub_nonprod_hubspot_connector_secret_values(
    envv: dict[str, str],
    *,
    database: str,
) -> dict[str, dict[str, int]]:
    """Remove only HubSpot connector secret values and prove the readback.

    Definitions stay in place so the next sanctioned deployment can provision
    an environment-scoped token.  No ciphertext or plaintext is recorded.
    """
    environment, target = verified_branch_reset_quarantine_target(envv)
    named_filter = _named_credential_filter()
    webhook_filter = _webhook_secret_filter()
    named_before = _scalar_count(
        db_query_scalar(envv, f"SELECT COUNT(*) {named_filter};", database=database),
        subject="HubSpot named credential secret",
    )
    webhook_before = _scalar_count(
        db_query_scalar(envv, f"SELECT COUNT(*) {webhook_filter};", database=database),
        subject="HubSpot webhook secret value",
    )
    # TiDB supports the MySQL multi-table DELETE form.  Each predicate binds
    # the deletion to the HubSpot connector's owned secret definitions.
    db_exec(
        envv,
        "DELETE secret FROM _System_NamedCredentialSecret secret "
        "JOIN _System_NamedCredential credential "
        "ON credential.Id = secret.NamedCredentialId "
        f"WHERE credential.DeveloperName = '{HUBSPOT_NAMED_CREDENTIAL}' "
        f"AND credential.OwnerPack = '{HUBSPOT_CONNECTOR_OWNER_PACK}' "
        "AND secret.SecretKind = 'Token';",
        database=database,
    )
    db_exec(
        envv,
        "DELETE value FROM _System_WebhookSecretValue value "
        "JOIN _System_WebhookSecret secret "
        "ON secret.Id = value.WebhookSecretId "
        f"WHERE secret.OwnerPack = '{HUBSPOT_CONNECTOR_OWNER_PACK}' "
        "AND secret.Name IN ('webhook', 'workflow-action') "
        "AND COALESCE(secret.IsDeleted, 0) = 0;",
        database=database,
    )
    named_remaining = _scalar_count(
        db_query_scalar(envv, f"SELECT COUNT(*) {named_filter};", database=database),
        subject="HubSpot named credential secret readback",
    )
    webhook_remaining = _scalar_count(
        db_query_scalar(envv, f"SELECT COUNT(*) {webhook_filter};", database=database),
        subject="HubSpot webhook secret value readback",
    )
    if named_remaining != 0 or webhook_remaining != 0:
        raise DeployError(
            "non-production HubSpot credential quarantine readback was not empty"
        )
    summary = {
        "_System_NamedCredentialSecret": {
            "deleted": named_before,
            "remaining": named_remaining,
        },
        "_System_WebhookSecretValue": {
            "deleted": webhook_before,
            "remaining": webhook_remaining,
        },
    }
    log(
        "quarantined non-production HubSpot connector secret values: "
        f"environment={environment} target={target} "
        f"namedCredentialDeleted={named_before} webhookDeleted={webhook_before}"
    )
    return summary


def verify_nonprod_hubspot_runtime_credential(
    envv: dict[str, str],
    *,
    target_environment: str,
    runtime_env_file: Path,
) -> dict[str, str]:
    """Read and verify the active runtime token without mutating the database."""
    runtime_environment = nonprod_runtime_environment(envv)
    selected_target = str(target_environment).strip().lower()
    expected_public_base_url = NONPROD_HUBSPOT_PUBLIC_BASE_URLS.get(selected_target)
    if expected_public_base_url is None:
        raise DeployError(
            f"unsupported non-production HubSpot runtime target: {selected_target}"
        )

    runtime_group, runtime_target = verified_branch_reset_quarantine_target(envv)
    expected_group = "staging" if selected_target == "staging" else "test"
    if runtime_group != expected_group:
        raise DeployError(
            f"selected target {selected_target} does not match runtime environment "
            f"{runtime_environment}"
        )
    if expected_group == "test" and runtime_target != selected_target:
        raise DeployError(
            f"selected target {selected_target} does not match runtime target "
            f"{runtime_target}"
        )

    envs_root = Path(
        os.environ.get("OFFLINE_ENVS_ROOT")
        or Path.home() / "nexuscrm-deployment" / "offline" / "envs"
    ).expanduser().resolve()
    expected_runtime_env_file = (
        envs_root / expected_group / runtime_target / "env" / "runtime.env"
    )
    requested_runtime_env_file = runtime_env_file.expanduser().absolute()
    if requested_runtime_env_file != expected_runtime_env_file:
        raise DeployError(
            "selected HubSpot runtime env path does not match the committed target: "
            f"expected {expected_runtime_env_file}, got {requested_runtime_env_file}"
        )
    expected_env_root = expected_runtime_env_file.parents[1]
    declared_env_root_raw = str(envv.get("ENV_ROOT", "")).strip()
    if not declared_env_root_raw:
        raise DeployError("selected HubSpot runtime env is missing ENV_ROOT")
    declared_env_root = Path(declared_env_root_raw).expanduser().resolve()
    if declared_env_root != expected_env_root:
        raise DeployError(
            "selected HubSpot runtime ENV_ROOT does not match the committed target"
        )

    target_config = topology_target_config(
        runtime_group,
        runtime_target,
        error_type=DeployError,
    )
    expected_backend_port = str(target_config.get("backend_port", "")).strip()
    if (
        not expected_backend_port
        or str(envv.get("BACKEND_PORT", "")).strip() != expected_backend_port
    ):
        raise DeployError(
            "selected HubSpot runtime backend port does not match the committed target"
        )
    if str(envv.get("PUBLIC_BASE_URL", "")).strip() != expected_public_base_url:
        raise DeployError(
            "selected HubSpot runtime public URL does not match the requested target"
        )

    database = _database(envv)
    access_token = _active_hubspot_access_token(envv, database)
    if access_token is None:
        raise DeployError(
            "non-production HubSpot runtime credential verification requires "
            "one active hubspot-crm-connector/HubSpotPrivateApp Token secret"
        )
    expected_identity = expected_nonprod_hubspot_identity(runtime_environment)
    identity = hubspot_account_identity(access_token)
    if identity != expected_identity:
        raise DeployError(
            "non-production HubSpot runtime credential identity mismatch: "
            f"expected portalId={expected_identity['portalId']} "
            f"accountType={expected_identity['accountType']}; "
            f"got portalId={identity['portalId']} accountType={identity['accountType']}"
        )
    log(
        "verified non-production HubSpot runtime credential: "
        f"environment={runtime_environment} portalId={identity['portalId']} "
        f"accountType={identity['accountType']}"
    )
    return identity


def enforce_hubspot_credential_admission(envv: dict[str, str]) -> dict[str, str] | None:
    runtime_environment = nonprod_runtime_environment(envv)
    database = _database(envv)
    try:
        access_token = _active_hubspot_access_token(envv, database)
        if access_token is None:
            if _active_hubspot_webhook_secret_count(envv, database) != 0:
                raise HubSpotCredentialQuarantineRequired(
                    "HubSpot credential admission found webhook secret values without an active Token"
                )
            log(
                "admitted non-production HubSpot startup without an active connector Token: "
                f"environment={runtime_environment}"
            )
            return None
        expected_identity = expected_nonprod_hubspot_identity(runtime_environment)
        identity = hubspot_account_identity(access_token)
        if identity != expected_identity:
            raise HubSpotCredentialQuarantineRequired(
                "non-production HubSpot credential identity does not match the committed policy"
            )
    except HubSpotCredentialQuarantineRequired as exc:
        try:
            scrub_nonprod_hubspot_connector_secret_values(envv, database=database)
        except DeployError as quarantine_exc:
            raise DeployError(
                "non-production HubSpot credential admission denied; "
                "backend startup is blocked and the database target was not eligible for automatic quarantine"
            ) from quarantine_exc
        raise DeployError(
            "non-production HubSpot credential admission denied; "
            "HubSpot connector secret values were quarantined and backend startup is blocked"
        ) from exc
    except DeployError as exc:
        raise DeployError(
            "non-production HubSpot credential admission denied; backend startup is blocked"
        ) from exc
    log(
        "admitted HubSpot credential for backend startup: "
        f"environment={runtime_environment} portalId={identity['portalId']} "
        f"accountType={identity['accountType']}"
    )
    return identity


def main() -> int:
    args = parse_args()
    try:
        envv = load_env_file(args.env_file)
        if args.verify_only:
            if not args.target_env:
                raise DeployError("--target-env is required with --verify-only")
            verify_nonprod_hubspot_runtime_credential(
                envv,
                target_environment=args.target_env,
                runtime_env_file=args.env_file,
            )
        else:
            enforce_hubspot_credential_admission(envv)
    except DeployError as exc:
        operation = "verification" if args.verify_only else "startup admission"
        log(f"HubSpot credential {operation} failed: {exc}")
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main())