Recopilar resultados de DomainTools Iris Investigate

En este documento se explica cómo ingerir resultados de DomainTools Iris Investigate en Google Security Operations mediante Amazon S3. El analizador transforma los datos JSON sin procesar de la API Iris de DomainTools en un formato estructurado que se ajusta al modelo de datos unificado (UDM) de Google SecOps. Extrae información relacionada con los detalles del dominio, la información de contacto, los riesgos de seguridad, los certificados SSL y otros atributos relevantes, y los asigna a los campos de UDM correspondientes para realizar análisis coherentes y obtener información sobre las amenazas.

Antes de empezar

• Instancia de Google SecOps
• Acceso privilegiado a la cuenta de empresa de DomainTools (acceso a la API de Iris Investigate)
• Acceso privilegiado a AWS (S3, IAM, Lambda y EventBridge)

Obtener la clave de API y el endpoint de DomainTools

1. Inicia sesión en el panel de control de la API de DomainTools (solo la cuenta del propietario de la API puede restablecer la clave de API).
2. En la sección Mi cuenta, seleccione el enlace Ver panel de APIs, situado en la pestaña Resumen de la cuenta.
3. Ve a la sección Nombre de usuario de la API para obtener tu nombre de usuario.
4. En la misma pestaña, busca tu clave de API.
5. Copia y guarda la clave en un lugar seguro.

6. Si necesitas una nueva clave, selecciona Restablecer clave de API.

7. Tenga en cuenta el endpoint de Iris Investigate: https://api.domaintools.com/v1/iris-investigate/.

Configurar un segmento de AWS S3 y IAM para Google SecOps

1. Crea un segmento de Amazon S3 siguiendo esta guía de usuario: Crear un segmento.
2. Guarda el nombre y la región del segmento para consultarlos más adelante (por ejemplo, domaintools-iris).
3. Crea un usuario siguiendo esta guía: Crear un usuario de gestión de identidades y accesos.
4. Selecciona el Usuario creado.
5. Selecciona la pestaña Credenciales de seguridad.
6. En la sección Claves de acceso, haz clic en Crear clave de acceso.
7. Selecciona Servicio de terceros como Caso práctico.
8. Haz clic en Siguiente.
9. Opcional: añade una etiqueta de descripción.
10. Haz clic en Crear clave de acceso.
11. Haz clic en Descargar archivo CSV para guardar la clave de acceso y la clave de acceso secreta para usarlas más adelante.
12. Haz clic en Listo.
13. Selecciona la pestaña Permisos.
14. En la sección Políticas de permisos, haz clic en Añadir permisos.
15. Selecciona Añadir permisos.
16. Seleccione Adjuntar políticas directamente.
17. Busca y selecciona la política AmazonS3FullAccess.
18. Haz clic en Siguiente.
19. Haz clic en Añadir permisos.

Configurar la política y el rol de gestión de identidades y accesos para las subidas de S3

1. En la consola de AWS, ve a IAM > Policies > Create policy > pestaña JSON.

2. Introduce la siguiente política:

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Sid": "AllowPutDomainToolsIrisObjects",
         "Effect": "Allow",
         "Action": "s3:PutObject",
         "Resource": "arn:aws:s3:::domaintools-iris/*"
       }
     ]
   }
   

   • Sustituye domaintools-iris si has introducido otro nombre de segmento.

3. Haz clic en Siguiente > Crear política.

4. Ve a IAM > Roles > Crear rol > Servicio de AWS > Lambda.

5. Adjunte la política que acaba de crear.

6. Dale el nombre WriteDomainToolsIrisToS3Role al rol y haz clic en Crear rol.

Crear la función Lambda

1. En la consola de AWS, ve a Lambda > Funciones > Crear función.
2. Haz clic en Crear desde cero.

3. Proporciona los siguientes detalles de configuración:

   Ajuste	Valor
   Nombre	domaintools_iris_to_s3
   Tiempo de ejecución	Python 3.13
   Arquitectura	x86_64
   Rol de ejecución	WriteDomainToolsIrisToS3Role

4. Una vez creada la función, abra la pestaña Código, elimine el stub e introduzca el siguiente código (domaintools_iris_to_s3.py):

   #!/usr/bin/env python3
   # Lambda: Pull DomainTools Iris Investigate results to S3 (no transform)
   
   import os, json, time, urllib.parse
   from urllib.request import Request, urlopen
   from urllib.error import HTTPError
   import boto3
   
   # --- Environment ---
   S3_BUCKET    = os.environ["S3_BUCKET"].strip()
   S3_PREFIX    = os.environ.get("S3_PREFIX", "domaintools/iris/").strip()
   STATE_KEY    = os.environ.get("STATE_KEY", "domaintools/iris/state.json").strip()
   
   DT_API_KEY   = os.environ["DT_API_KEY"].strip()
   DT_API_SECRET= os.environ.get("DT_API_SECRET", "").strip()  # optional if your account uses key-only auth
   
   USE_MODE     = os.environ.get("USE_MODE", "HASH").strip().upper()  # HASH | DOMAINS | QUERY
   SEARCH_HASHES= [h.strip() for h in os.environ.get("SEARCH_HASHES", "").split(";") if h.strip()]
   DOMAINS      = [d.strip() for d in os.environ.get("DOMAINS", "").split(";") if d.strip()]
   QUERY_LIST   = [q.strip() for q in os.environ.get("QUERY_LIST", "").split(";") if q.strip()]
   
   PAGE_SIZE    = int(os.environ.get("PAGE_SIZE", "500"))
   MAX_PAGES    = int(os.environ.get("MAX_PAGES", "20"))
   USE_NEXT     = os.environ.get("USE_NEXT", "true").lower() == "true"
   HTTP_TIMEOUT = int(os.environ.get("HTTP_TIMEOUT", "60"))
   RETRIES      = int(os.environ.get("HTTP_RETRIES", "2"))
   
   BASE_URL = "https://api.domaintools.com/v1/iris-investigate/"
   HDRS = {
       "X-Api-Key": DT_API_KEY,
       "Accept": "application/json",
   }
   if DT_API_SECRET:
       HDRS["X-Api-Secret"] = DT_API_SECRET
   
   s3 = boto3.client("s3")
   
   # --- HTTP helpers ---
   
   def _http_get(url: str) -> dict:
       req = Request(url, method="GET")
       for k, v in HDRS.items():
           req.add_header(k, v)
       attempt = 0
       while True:
           try:
               with urlopen(req, timeout=HTTP_TIMEOUT) as r:
                   return json.loads(r.read().decode("utf-8"))
           except HTTPError as e:
               if e.code in (429, 500, 502, 503, 504) and attempt < RETRIES:
                   delay = int(e.headers.get("Retry-After", "2"))
                   time.sleep(max(1, delay))
                   attempt += 1
                   continue
               raise
   
   def _build_url(params: dict) -> str:
       return BASE_URL + ("?" + urllib.parse.urlencode(params, doseq=True) if params else "")
   
   # --- S3 helpers ---
   
   def _write_page(obj: dict, label: str, page: int) -> str:
       ts = time.strftime("%Y/%m/%d/%H%M%S", time.gmtime())
       key = f"{S3_PREFIX.rstrip('/')}/{ts}-{label}-p{page:05d}.json"
       s3.put_object(
           Bucket=S3_BUCKET, Key=key,
           Body=json.dumps(obj, separators=(",", ":")).encode("utf-8"),
           ContentType="application/json",
       )
       return key
   
   # --- Iris paging ---
   
   def _first_page_params() -> dict:
       params: dict[str, object] = {"page_size": str(PAGE_SIZE)}
       if USE_NEXT:
           params["next"] = "true"
       return params
   
   def _paginate(label: str, params: dict) -> tuple[int, int]:
       pages = 0
       total = 0
       url = _build_url(params)
       while pages < MAX_PAGES:
           data = _http_get(url)
           _write_page(data, label, pages)
   
           resp = data.get("response") or {}
           results = resp.get("results") or []
           total += len(results)
           pages += 1
   
           # Prefer `next` absolute URL if present
           next_url = resp.get("next") if isinstance(resp, dict) else None
           if next_url:
               url = next_url
               continue
   
           # Fallback: position pager when `next=true` not used/supported
           if resp.get("has_more_results") and resp.get("position"):
               base = _first_page_params()
               base.pop("next", None)
               base["position"] = resp["position"]
               url = _build_url(base)
               continue
   
           break
       return pages, total
   
   # --- Mode runners ---
   
   def run_hashes(hashes: list[str]) -> dict:
       agg_pages = agg_results = 0
       for h in hashes:
           params = _first_page_params()
           params["search_hash"] = h
           p, r = _paginate(f"hash-{h}", params)
           agg_pages += p
           agg_results += r
       return {"pages": agg_pages, "results": agg_results}
   
   def run_domains(domains: list[str]) -> dict:
       agg_pages = agg_results = 0
       for d in domains:
           params = _first_page_params()
           # DomainTools accepts `domain` as a filter in Investigate search
           params["domain"] = d
           p, r = _paginate(f"domain-{d}", params)
           agg_pages += p
           agg_results += r
       return {"pages": agg_pages, "results": agg_results}
   
   def run_queries(queries: list[str]) -> dict:
       agg_pages = agg_results = 0
       for q in queries:
           # Merge arbitrary k=v pairs from the query string
           base = _first_page_params()
           for k, v in urllib.parse.parse_qsl(q, keep_blank_values=True):
               base.setdefault(k, v)
           p, r = _paginate(f"query-{q.replace('=','-')}", base)
           agg_pages += p
           agg_results += r
       return {"pages": agg_pages, "results": agg_results}
   
   # --- Entry point ---
   
   def lambda_handler(event=None, context=None):
       if USE_MODE == "HASH" and SEARCH_HASHES:
           res = run_hashes(SEARCH_HASHES)
       elif USE_MODE == "DOMAINS" and DOMAINS:
           res = run_domains(DOMAINS)
       elif USE_MODE == "QUERY" and QUERY_LIST:
           res = run_queries(QUERY_LIST)
       else:
           raise ValueError("Invalid USE_MODE or missing parameters. Set USE_MODE to HASH | DOMAINS | QUERY and provide SEARCH_HASHES | DOMAINS | QUERY_LIST accordingly.")
   
       return {"ok": True, "mode": USE_MODE, **res}
   
   if __name__ == "__main__":
       print(json.dumps(lambda_handler(), indent=2))
   

5. Ve a Configuración > Variables de entorno > Editar > Añadir nueva variable de entorno.

6. Introduce las siguientes variables de entorno y sustituye los valores por los tuyos:

   Clave	Valor de ejemplo	Descripción
   S3_BUCKET	domaintools-iris	Nombre del segmento de S3 en el que se almacenarán los datos.
   S3_PREFIX	domaintools/iris/	Prefijo de S3 opcional (subcarpeta) para los objetos.
   STATE_KEY	domaintools/iris/state.json	Clave de archivo de estado o punto de control opcional.
   DT_API_KEY	DT-XXXXXXXXXXXXXXXXXXXX	Clave de API de DomainTools.
   DT_API_SECRET	YYYYYYYYYYYYYYYYYYYYYYYY	Secreto de la API de DomainTools (si procede).
   USE_MODE	HASH | DOMAINS | QUERY	Selecciona el modo que quieras usar (solo puede haber uno activo a la vez).
   SEARCH_HASHES	hash1;hash2;hash3	Obligatorio si USE_MODE=HASH. Lista separada por punto y coma de los hashes de las búsquedas guardadas de la interfaz de usuario de Iris.
   DOMAINS	example.com;domaintools.com	Obligatorio si USE_MODE=DOMAINS. Lista de dominios separados por punto y coma.
   QUERY_LIST	ip=1.1.1.1;ip=8.8.8.8;domain=example.org	Obligatorio si USE_MODE=QUERY. Lista de cadenas de consulta separadas por punto y coma (k=v&k2=v2).
   PAGE_SIZE	500	Filas por página (500 de forma predeterminada).
   MAX_PAGES	20	Número máximo de páginas por solicitud

7. Una vez creada la función, permanece en su página (o abre Lambda > Funciones > tu-función).

8. Seleccione la pestaña Configuración.

9. En el panel Configuración general, haz clic en Editar.

10. Cambia Tiempo de espera a 15 minutos (900 segundos) y haz clic en Guardar.

Crear una programación de EventBridge

1. Ve a Amazon EventBridge > Scheduler > Create schedule (Amazon EventBridge > Programador > Crear programación).
2. Proporcione los siguientes detalles de configuración:
   • Programación periódica: Precio (1 hour).
   • Destino: tu función Lambda.
   • Nombre: domaintools-iris-1h.
3. Haz clic en Crear programación.

Opcional: Crear un usuario y claves de gestión de identidades y accesos de solo lectura para Google SecOps

1. En la consola de AWS, vaya a IAM > Usuarios y, a continuación, haga clic en Añadir usuarios.
2. Proporcione los siguientes detalles de configuración:
   • Usuario: introduce un nombre único (por ejemplo, secops-reader).
   • Tipo de acceso: selecciona Clave de acceso - Acceso programático.
   • Haz clic en Crear usuario.
3. Asigna una política de lectura mínima (personalizada): Usuarios > selecciona secops-reader > Permisos > Añadir permisos > Asignar políticas directamente > Crear política.

4. En el editor de JSON, introduce la siguiente política:

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": ["s3:GetObject"],
         "Resource": "arn:aws:s3:::<your-bucket>/*"
       },
       {
         "Effect": "Allow",
         "Action": ["s3:ListBucket"],
         "Resource": "arn:aws:s3:::<your-bucket>"
       }
     ]
   }
   

5. Asigna el nombre secops-reader-policy.

6. Ve a Crear política > busca o selecciona > Siguiente > Añadir permisos.

7. Ve a Credenciales de seguridad > Claves de acceso > Crear clave de acceso.

8. Descarga el archivo CSV (estos valores se introducen en el feed).

Configurar un feed en Google SecOps para ingerir resultados de DomainTools Iris Investigate

1. Ve a Configuración de SIEM > Feeds.
2. Haz clic en Añadir feed.
3. En el campo Nombre del feed, introduce un nombre para el feed (por ejemplo, DomainTools Iris Investigate).
4. Selecciona Amazon S3 V2 como Tipo de fuente.
5. Seleccione DomainTools Threat Intelligence como Tipo de registro.
6. Haz clic en Siguiente.
7. Especifique los valores de los siguientes parámetros de entrada:
   • URI de S3: s3://domaintools-iris/domaintools/iris/
   • Opciones de eliminación de la fuente: selecciona la opción de eliminación que prefieras.
   • Antigüedad máxima de los archivos: 180 días de forma predeterminada.
   • ID de clave de acceso: clave de acceso de usuario con acceso al bucket de S3.
   • Clave de acceso secreta: clave secreta del usuario con acceso al bucket de S3.
   • Espacio de nombres del recurso: domaintools.threat_intel
   • Etiquetas de ingestión: etiqueta que se aplicará a los eventos de este feed.
8. Haz clic en Siguiente.
9. Revise la configuración de la nueva fuente en la pantalla Finalizar y, a continuación, haga clic en Enviar.

Tabla de asignación de UDM

¿Necesitas más ayuda? Recibe respuestas de los miembros de la comunidad y de los profesionales de Google SecOps.