# Copyright 2026 SigmaSRC, Inc. All Rights Reserved.

"""Gather DNS resolutions of a watchlist of AI-provider domains.

Cloud LLM APIs (OpenAI, Google Gemini, Cohere, ...) sit behind shared CDNs and
publish no dedicated IP range, so the server can't name them from a flow's
destination IP alone. This gather re-resolves a fixed list of AI-provider
hostnames *as the host's own resolver sees them* and reports
(domain -> resolved_ip). The server then correlates a network flow's dest IP to
the domain this host resolved and matches it to a provider (SigmaAIShield AI
Usage report).

Table columns:

- domain       -- the AI-provider hostname resolved
- resolved_ip  -- an address (A / AAAA) the host's resolver returned for it

Only fixed public API/web endpoints are covered. Per-tenant endpoints
(Azure OpenAI ``*.openai.azure.com``, AWS Bedrock ``bedrock-*.amazonaws.com``)
are customer-specific and can't be watchlisted here.
"""

import os
import sys
import socket
import logging

if __name__ == "__main__":
    try:
        import sigma
    except ImportError:
        sys.path.append(os.path.join(os.pardir, os.pardir))

logger = logging.getLogger("gather.dns")

_COLUMNS = ["domain", "resolved_ip"]

# Public AI-provider API/web endpoints. Keep in sync with the server's
# ai_egress_signature domain_pattern rows (migration 424-dns-cloud-app-id).
_WATCHLIST = [
    "api.openai.com", "chatgpt.com",
    "api.anthropic.com", "claude.ai",
    "generativelanguage.googleapis.com", "aiplatform.googleapis.com",
    "api.cohere.com", "api.mistral.ai", "api.perplexity.ai",
    "api-inference.huggingface.co", "api.x.ai", "api.groq.com",
    "api.deepseek.com", "api.githubcopilot.com",
]


def Gather(collector):
    t = collector.table("aidns", _COLUMNS)
    for domain in _WATCHLIST:
        try:
            infos = socket.getaddrinfo(domain, None)
        except (socket.gaierror, OSError):
            continue  # not resolvable from this host — skip
        seen = set()
        for info in infos:
            ip = info[4][0]
            if ip and ip not in seen:
                seen.add(ip)
                t.add([domain, ip])


def _test():
    from sigma.gather import gathering
    gathering.test(Gather)


if __name__ == "__main__":
    _test()
