"""AI-service whitelist resolver — drives the AI Spending dashboard's cost classification AND per-AK product attribution from Phase 2 on. Background ---------- PR #253 (Phase 2 of the AI Spending refactor) retired the ``cmdb_cloud_scopes.is_ai_dedicated`` boolean and the parallel ``cmdb_ai_resources`` ledger. AI cost is now identified exclusively by matching ``cmdb_cost_facts.service`` against a configurable per-provider whitelist stored in ``platform_settings.ai_spending.service_whitelist``. The whitelist is a JSON dict keyed by integration ``provider`` slug (``azure`` / ``aws`` / ``gcp``); each value is a list of substring patterns. Aliyun is deliberately excluded — Bailian (the only AI product the tenant bills on Aliyun) flows through ``bailian_usage_daily`` as a first-class bucket; treating any Aliyun service as "cloud AI" would double-count Bailian. Matching semantics ------------------ * Case-insensitive substring (``pattern.lower() in service.lower()``) * Empty / whitespace-only patterns are skipped * Unknown providers => no match (graceful no-op; lets the tenant enable a new cloud without immediately falsing AI cost classification) This module is the SINGLE source of truth — every call site (overview aggregation, per-AK spend, breakdown, MCP tools) MUST route through ``service_matches_ai_for_provider`` or ``build_service_filter_sql`` so admin-edited patterns propagate everywhere on the next request. """ from __future__ import annotations import logging from typing import Final from sqlalchemy import or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.integrations.models import PlatformSetting logger = logging.getLogger(__name__) SETTING_KEY: Final[str] = "ai_spending.service_whitelist" # Provider slugs the whitelist may key on. Matches ``integrations.provider`` # values for ``kind='cloud'`` rows that host first-class AI services. # Adding a new cloud here is admin-config only (whitelist value edit); # this list is the authoritative set of recognised keys so that a # typo'd whitelist key (``"azuer": [...]``) is logged loudly rather # than silently disabling matching for that cloud. SUPPORTED_PROVIDERS: Final[frozenset[str]] = frozenset( {"azure", "aws", "gcp"}, ) # Maximum patterns per provider — admin error guard. A whitelist # with 1000+ substring patterns would OR-fold into a query plan TiDB # refuses to optimise; we surface the limit at validate-write time # rather than discovering it at slow-query time. MAX_PATTERNS_PER_PROVIDER: Final[int] = 64 # Per-pattern length cap — same defensive ceiling on admin input. MAX_PATTERN_LENGTH: Final[int] = 128 # Per-pattern length FLOOR — substring matching with pattern length 1-3 # overmatches dramatically: ``"AI"`` matches every service whose name # contains those two letters (AvailableInstances, AKS, RemoteAccessIaaS # would all false-match into a Foundry bucket). Require at least 4 # characters to make the substring match meaningful. Common AI product # names (Bedrock, Foundry, Cursor, Codex, Claude, OpenAI, Gemini, # Anthropic, Mistral, Cohere, Llama) all clear the floor. MIN_PATTERN_LENGTH: Final[int] = 4 async def load_whitelist(db: AsyncSession) -> dict[str, list[str]]: """Return the current whitelist dict. Reads ``platform_settings`` at request time so admin edits take effect on the next call — no app restart, no cache invalidation hooks. The row is a single small JSON document; the read is sub-millisecond. Raises a ``RuntimeError`` if the row is missing — the platform_setting is seeded by migration 149 and must exist on any deployment that's been migrated past that revision. A missing row indicates a partially applied migration chain and we refuse to operate (rather than silently treat every cost row as non-AI, which would zero out the dashboard). Returns an empty dict when the whitelist exists but is empty/blank; callers downstream produce $0 totals in that case (an admin who has deliberately blanked the whitelist gets an honest empty AI dashboard, not a guess based on dead defaults). """ row = ( await db.execute( select(PlatformSetting).where( PlatformSetting.key == SETTING_KEY, ) ) ).scalar_one_or_none() if row is None: raise RuntimeError( f"Required platform_setting {SETTING_KEY!r} is missing — " "run alembic upgrade to apply migration 149.", ) value = row.value if not isinstance(value, dict): raise RuntimeError( f"platform_setting {SETTING_KEY!r} must be a JSON object, " f"got {type(value).__name__}: {value!r}", ) # Normalise: strip empty entries, lowercase the key (provider slug # is already lower in integrations.provider — defensive), and warn # about unknown keys. out: dict[str, list[str]] = {} for raw_key, raw_patterns in value.items(): key = str(raw_key).strip().lower() if not key: continue if key not in SUPPORTED_PROVIDERS: logger.warning( "ai_spending.whitelist: unknown provider key %r — " "ignored. Recognised keys: %s", raw_key, ", ".join(sorted(SUPPORTED_PROVIDERS)), ) continue if not isinstance(raw_patterns, list): logger.warning( "ai_spending.whitelist: provider %r value is not a " "list (got %s), ignored.", key, type(raw_patterns).__name__, ) continue patterns = [ str(p).strip() for p in raw_patterns if isinstance(p, str) and str(p).strip() ] out[key] = patterns return out def validate_whitelist_payload(value: object) -> dict[str, list[str]]: """Validate admin-supplied whitelist before persisting. Mirrors :func:`load_whitelist` shape but is strict about errors — the settings router calls this to surface ``400 Bad Request`` with a specific complaint instead of silently dropping malformed input on the next read. Validations: * Top-level must be a JSON object. * Keys must be in :data:`SUPPORTED_PROVIDERS` (lowercased). * Values must be lists of strings. * Each pattern <= :data:`MAX_PATTERN_LENGTH` chars after strip. * <= :data:`MAX_PATTERNS_PER_PROVIDER` patterns per provider. * Empty patterns silently dropped (admin convenience). Returns the normalised dict ready to persist; never returns a dict that ``load_whitelist`` would warn about. """ if not isinstance(value, dict): raise ValueError( f"whitelist must be a JSON object, got " f"{type(value).__name__}", ) out: dict[str, list[str]] = {} for raw_key, raw_patterns in value.items(): if not isinstance(raw_key, str): raise ValueError( f"provider key must be a string, got " f"{type(raw_key).__name__}: {raw_key!r}", ) key = raw_key.strip().lower() if not key: continue if key not in SUPPORTED_PROVIDERS: raise ValueError( f"unknown provider {key!r}; supported: " f"{sorted(SUPPORTED_PROVIDERS)}", ) if not isinstance(raw_patterns, list): raise ValueError( f"provider {key!r} value must be a list, got " f"{type(raw_patterns).__name__}", ) patterns: list[str] = [] for p in raw_patterns: if not isinstance(p, str): raise ValueError( f"provider {key!r}: every pattern must be a " f"string, got {type(p).__name__}: {p!r}", ) trimmed = p.strip() if not trimmed: continue if len(trimmed) < MIN_PATTERN_LENGTH: raise ValueError( f"provider {key!r}: pattern length must be " f">= {MIN_PATTERN_LENGTH} (substring match on " f"shorter tokens over-matches unrelated services), " f"got {len(trimmed)} for {trimmed!r}", ) if len(trimmed) > MAX_PATTERN_LENGTH: raise ValueError( f"provider {key!r}: pattern length must be " f"<= {MAX_PATTERN_LENGTH}, got {len(trimmed)} for " f"{trimmed[:32]!r}...", ) patterns.append(trimmed) if len(patterns) > MAX_PATTERNS_PER_PROVIDER: raise ValueError( f"provider {key!r}: at most " f"{MAX_PATTERNS_PER_PROVIDER} patterns allowed, got " f"{len(patterns)}", ) out[key] = patterns return out def service_matches_ai_for_provider( service: str | None, patterns: list[str], ) -> bool: """True iff ``service`` contains any of ``patterns`` (case-insensitive). ``service`` is the ``cmdb_cost_facts.service`` value for a single fact row; ``patterns`` is the whitelist list for that fact's integration provider. Empty / None ``service`` always fails — a cost-fact row with no service classification is not attributable to any AI product. """ if not service or not patterns: return False svc_lower = service.lower() return any( p.lower() in svc_lower for p in patterns if p and p.strip() ) def build_service_filter_sql(column, patterns: list[str]): """Return a SQLAlchemy expression: ``column ILIKE %pattern%`` joined with OR for every non-empty pattern. ``column`` is the SQLAlchemy column reference (typically ``CmdbCostFact.service``). The caller is responsible for combining this with the per-integration filter (``integration_id == X``) so that an Azure ``Foundry`` pattern doesn't accidentally match an AWS service named ``foundry-something``. Returns an always-false expression when ``patterns`` is empty — produces no rows, no exception. Matches the in-memory helper's behaviour: an unknown / empty provider list excludes all rows rather than including them. """ clean = [p for p in patterns if p and p.strip()] if not clean: # ``column.is_(None) & column.isnot(None)`` is always-false in # SQL. The optimiser short-circuits it, so no rows scan. return column.is_(None) & column.isnot(None) # Escape LIKE metacharacters so an admin-supplied pattern containing # ``%`` or ``_`` matches them literally rather than as wildcards. def _esc(p: str) -> str: return p.replace("\\", "\\\\").replace("%", r"\%").replace("_", r"\_") return or_(*(column.ilike(f"%{_esc(p)}%", escape="\\") for p in clean)) __all__ = [ "MAX_PATTERNS_PER_PROVIDER", "MAX_PATTERN_LENGTH", "SETTING_KEY", "SUPPORTED_PROVIDERS", "build_service_filter_sql", "load_whitelist", "service_matches_ai_for_provider", "validate_whitelist_payload", ]