Recopila los resultados de DomainTools Iris Investigate

En este documento, se explica cómo transferir los resultados de DomainTools Iris Investigate a Google Security Operations con Amazon S3. El analizador transforma los datos JSON sin procesar de la API de 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 correspondientes del UDM para un análisis coherente y la inteligencia sobre amenazas.

Antes de comenzar

• Instancia de Google SecOps
• Acceso con privilegios a la cuenta empresarial de DomainTools (acceso a la API de Iris Investigate)
• Acceso con privilegios a AWS (S3, IAM, Lambda, EventBridge)

Obtén la clave de API y el extremo de DomainTools

1. Accede al panel 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, selecciona el vínculo Ver panel de la API ubicado 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 una ubicación segura.

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

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

Configura el bucket de AWS S3 y el IAM para Google SecOps

1. Crea un bucket de Amazon S3 siguiendo esta guía del usuario: Crea un bucket
2. Guarda el Nombre y la Región del bucket para futuras referencias (por ejemplo, domaintools-iris).
3. Crea un usuario siguiendo esta guía del usuario: Cómo crear un usuario de IAM.
4. Selecciona el usuario creado.
5. Selecciona la pestaña Credenciales de seguridad.
6. Haz clic en Crear clave de acceso en la sección Claves de acceso.
7. Selecciona Servicio de terceros como el Caso de uso.
8. Haz clic en Siguiente.
9. Opcional: Agrega 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. Haz clic en Agregar permisos en la sección Políticas de permisos.
15. Selecciona Agregar permisos.
16. Selecciona Adjuntar políticas directamente.
17. Busca y selecciona la política AmazonS3FullAccess.
18. Haz clic en Siguiente.
19. Haz clic en Agregar permisos.

Configura la política y el rol de IAM para las cargas de S3

1. En la consola de AWS, ve a IAM > Políticas > Crear política > pestaña JSON.

2. Ingresa la siguiente política:

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

   • Reemplaza domaintools-iris si ingresaste un nombre de bucket diferente.

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

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

5. Adjunta la política recién creada.

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

Crea la función Lambda

1. En la consola de AWS, ve a Lambda > Functions > Create function.
2. Haz clic en Crear desde cero.

3. Proporciona los siguientes detalles de configuración:

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

4. Después de crear la función, abre la pestaña Code, borra el código auxiliar y, luego, ingresa 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 > Agregar nueva variable de entorno.

6. Ingresa las siguientes variables de entorno y reemplaza los valores por los tuyos:

   Clave	Valor de ejemplo	Descripción
   S3_BUCKET	domaintools-iris	Nombre del bucket 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	Es la clave opcional del archivo de estado o punto de control.
   DT_API_KEY	DT-XXXXXXXXXXXXXXXXXXXX	Es la clave de API de DomainTools.
   DT_API_SECRET	YYYYYYYYYYYYYYYYYYYYYYYY	Secreto de la API de DomainTools (si corresponde)
   USE_MODE	HASH | DOMAINS | QUERY	Selecciona el modo que quieres usar (solo uno está activo a la vez).
   SEARCH_HASHES	hash1;hash2;hash3	Obligatorio si USE_MODE=HASH. Lista de hashes de búsquedas guardadas separados por punto y coma desde la IU de Iris.
   DOMAINS	example.com;domaintools.com	Obligatorio si USE_MODE=DOMAINS. Es una 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 (el valor predeterminado es 500).
   MAX_PAGES	20	Cantidad máxima de páginas por solicitud

7. Después de crear la función, permanece en su página (o abre Lambda > Functions > tu-función).

8. Selecciona 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.

Crea una programación de EventBridge

1. Ve a Amazon EventBridge > Scheduler > Create schedule.
2. Proporciona los siguientes detalles de configuración:
   • Programación recurrente: Frecuencia (1 hour).
   • Destino: Tu función Lambda.
   • Nombre: domaintools-iris-1h.
3. Haz clic en Crear programación.

Opcional: Crea un usuario y claves de IAM de solo lectura para Google SecOps

1. En la consola de AWS, ve a IAM > Usuarios y, luego, haz clic en Agregar usuarios.
2. Proporciona los siguientes detalles de configuración:
   • Usuario: Ingresa un nombre único (por ejemplo, secops-reader).
   • Tipo de acceso: Selecciona Clave de acceso: Acceso programático.
   • Haz clic en Crear usuario.
3. Adjunta una política de lectura mínima (personalizada): Usuarios > selecciona secops-reader > Permisos > Agregar permisos > Adjuntar políticas directamente > Crear política

4. En el editor de JSON, ingresa 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. Configura el nombre como secops-reader-policy.

6. Ve a Crear política > busca o selecciona > Siguiente > Agregar permisos.

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

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

Configura un feed en Google SecOps para transferir los resultados de DomainTools Iris Investigate

1. Ve a SIEM Settings > Feeds.
2. Haz clic en Agregar feed nuevo.
3. En el campo Nombre del feed, ingresa un nombre para el feed (por ejemplo, DomainTools Iris Investigate).
4. Selecciona Amazon S3 V2 como el Tipo de fuente.
5. Selecciona DomainTools Threat Intelligence como el Tipo de registro.
6. Haz clic en Siguiente.
7. Especifica valores para los siguientes parámetros de entrada:
   • URI de S3: s3://domaintools-iris/domaintools/iris/
   • Opciones de borrado de la fuente: Selecciona la opción de borrado según tu preferencia.
   • Antigüedad máxima del archivo: 180 días de forma predeterminada
   • ID de clave de acceso: Clave de acceso del 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 activo: domaintools.threat_intel
   • Etiquetas de transmisión: Es la etiqueta que se aplicará a los eventos de este feed.
8. Haz clic en Siguiente.
9. Revisa la nueva configuración del feed en la pantalla Finalizar y, luego, haz clic en Enviar.

Tabla de asignación de UDM

¿Necesitas más ayuda? Obtén respuestas de miembros de la comunidad y profesionales de Google SecOps.