Collect DomainTools Iris Investigate results

This document explains how to ingest DomainTools Iris Investigate results to Google Security Operations using Amazon S3. The parser transforms raw JSON data from DomainTools' Iris API into a structured format conforming to Google SecOps's Unified Data Model (UDM). It extracts information related to domain details, contact information, security risks, SSL certificates, and other relevant attributes, mapping them to corresponding UDM fields for consistent analysis and threat intelligence.

Before you begin

• Google SecOps instance
• Privileged access to DomainTools enterprise account (API access to Iris Investigate)
• Privileged access to AWS (S3, IAM, Lambda, EventBridge)

Get DomainTools API Key and Endpoint

1. Sign in to the DomainTools API Dashboard (only the API owner account can reset the API key).
2. In the My Account section, select the View API Dashboard link located in the Account Summary tab.
3. Go to the API Username section to obtain your username.
4. In the same tab, locate your API Key.
5. Copy and save the key in a secure location.

6. If you need a new key, select Reset API Key.

7. Note the Iris Investigate endpoint: https://api.domaintools.com/v1/iris-investigate/.

Configure AWS S3 bucket and IAM for Google SecOps

1. Create Amazon S3 bucket following this user guide: Creating a bucket
2. Save bucket Name and Region for future reference (for example, domaintools-iris).
3. Create a user following this user guide: Creating an IAM user.
4. Select the created User.
5. Select the Security credentials tab.
6. Click Create Access Key in the Access Keys section.
7. Select Third-party service as the Use case.
8. Click Next.
9. Optional: add a description tag.
10. Click Create access key.
11. Click Download CSV file to save the Access Key and Secret Access Key for later use.
12. Click Done.
13. Select the Permissions tab.
14. Click Add permissions in the Permissions policies section.
15. Select Add permissions.
16. Select Attach policies directly
17. Search for and select the AmazonS3FullAccess policy.
18. Click Next.
19. Click Add permissions.

Configure the IAM policy and role for S3 uploads

1. In the AWS console, go to IAM > Policies > Create policy > JSON tab.

2. Enter the following policy:

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

   • Replace domaintools-iris if you entered a different bucket name.

3. Click Next > Create policy.

4. Go to IAM > Roles > Create role > AWS service > Lambda.

5. Attach the newly created policy.

6. Name the role WriteDomainToolsIrisToS3Role and click Create role.

Create the Lambda function

1. In the AWS Console, go to Lambda > Functions > Create function.
2. Click Author from scratch.

3. Provide the following configuration details:

   Setting	Value
   Name	domaintools_iris_to_s3
   Runtime	Python 3.13
   Architecture	x86_64
   Execution role	WriteDomainToolsIrisToS3Role

4. After the function is created, open the Code tab, delete the stub and enter the following code (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. Go to Configuration > Environment variables > Edit > Add new environment variable.

6. Enter the following environment variables, replacing with your values:

   Key	Example Value	Description
   S3_BUCKET	domaintools-iris	S3 bucket name where data will be stored.
   S3_PREFIX	domaintools/iris/	Optional S3 prefix (subfolder) for objects.
   STATE_KEY	domaintools/iris/state.json	Optional state/checkpoint file key.
   DT_API_KEY	DT-XXXXXXXXXXXXXXXXXXXX	DomainTools API key.
   DT_API_SECRET	YYYYYYYYYYYYYYYYYYYYYYYY	DomainTools API secret (if applicable).
   USE_MODE	HASH | DOMAINS | QUERY	Select which mode to use (only one is active at a time).
   SEARCH_HASHES	hash1;hash2;hash3	Required if USE_MODE=HASH. Semicolon-separated list of saved search hashes from the Iris UI.
   DOMAINS	example.com;domaintools.com	Required if USE_MODE=DOMAINS. Semicolon-separated list of domains.
   QUERY_LIST	ip=1.1.1.1;ip=8.8.8.8;domain=example.org	Required if USE_MODE=QUERY. Semicolon-separated list of query strings (k=v&k2=v2).
   PAGE_SIZE	500	Rows per page (default 500).
   MAX_PAGES	20	Max pages per request

7. After the function is created, stay on its page (or open Lambda > Functions > your-function).

8. Select the Configuration tab.

9. In the General configuration panel, click Edit.

10. Change Timeout to 15 minutes (900 seconds) and click Save.

Create an EventBridge schedule

1. Go to Amazon EventBridge > Scheduler > Create schedule.
2. Provide the following configuration details:
   • Recurring schedule: Rate (1 hour).
   • Target: your Lambda function.
   • Name: domaintools-iris-1h.
3. Click Create schedule.

Optional: Create read-only IAM user & keys for Google SecOps

1. In the AWS Console, go to IAM > Users, then click Add users.
2. Provide the following configuration details:
   • User: Enter a unique name (for example, secops-reader)
   • Access type: Select Access key - Programmatic access
   • Click Create user.
3. Attach minimal read policy (custom): Users > select secops-reader > Permissions > Add permissions > Attach policies directly > Create policy

4. In the JSON editor, enter the following policy:

   {
     "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. Set the name to secops-reader-policy.

6. Go to Create policy > search/select > Next > Add permissions.

7. Go to Security credentials > Access keys > Create access key.

8. Download the CSV (these values are entered into the feed).

Configure a feed in Google SecOps to ingest DomainTools Iris Investigate results

1. Go to SIEM Settings > Feeds.
2. Click Add New Feed.
3. In the Feed name field, enter a name for the feed (for example, DomainTools Iris Investigate).
4. Select Amazon S3 V2 as the Source type.
5. Select DomainTools Threat Intelligence as the Log type.
6. Click Next.
7. Specify values for the following input parameters:
   • S3 URI: s3://domaintools-iris/domaintools/iris/
   • Source deletion options: Select the deletion option according to your preference.
   • Maximum File Age: Default 180 Days.
   • Access Key ID: User access key with access to the S3 bucket.
   • Secret Access Key: User secret key with access to the S3 bucket.
   • Asset namespace: domaintools.threat_intel
   • Ingestion labels: The label to be applied to the events from this feed.
8. Click Next.
9. Review your new feed configuration in the Finalize screen, and then click Submit.

UDM Mapping Table

Need more help? Get answers from Community members and Google SecOps professionals.