Threat Intelligence API

Who it's for

Threat-intel and ASM platforms, security teams automating recon, and OSINT products that need search data without a scraping layer.

What you get

• Full SERPs for dork-style queries (filetype:, site:, inurl:) as structured data
• News and search coverage for phishing kit and impersonation signals
• What AI engines say when asked about a domain or brand
• Webhook-delivered async tasks for watch lists that run continuously

Try it

curl -X POST "https://api.answerline.dev/v1/monitor/google" \
  -H "Authorization: Bearer $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"query":"site:acme-secure-login.com OR inurl:acme-verify","country":"US"}'

Sync or queue it with POST /v1/async/task. Full field list on the Google Search page and in the API reference.

Engines for this workload

Open-web reconnaissance as a feed

Much external threat intelligence is reading public indexes on a loop. Is a new login page using our name? Has a document from our domain been indexed that should not be? Is a phishing campaign against our customers being reported? Analysts answer these with search operators and news searches, and doing it by hand makes coverage depend on who ran which query when.

Through the API, every query returns a parsed page: organicResults with title, link, displayedLink, snippet and date from Google Search, newsResults from Google News, and text and sources from AI engines such as ChatGPT. Your pipeline turns repeated results into first-seen events.

The data is what the public Google index and AI answers showed for a query at a moment. The API does not crawl, scan or access the linked pages, and a result is a lead for an analyst, not a confirmed threat. Only monitor assets you are responsible for or authorized to assess, and handle found URLs with the care you give any untrusted link.

Metrics

Architecture

assets (domains, brands, products, key strings)
      │
template library (dork templates × asset) ──> scheduler ──> batches ──> webhook
                                                                        │
                           results (url, first_seen, last_seen) ◄── normalize + diff
                                       │
                 classifier ──> alerts (SIEM / ticket) ──> analyst verdicts ──> template tuning

Asset inventory

Start from what you own: registered domains and subdomains, brand and product names, distinctive strings from your login pages (page titles, footer text), public code or package names, and support email domains. Give each asset an id and a list of templates that apply to it.

Template library

Templates are parameterized queries grouped by what they detect. Examples with a placeholder asset:

Keep templates in data with an owner, a severity and an expected yield. Test each template manually before scheduling it, because operator behavior is Google’s and can change. To go past the first results, use pages rather than a result offset: a complete Google search URL passed as url instead of query may carry only q, gl, num, hl and uule, not start, and is not combinable with pages.

Schedule

Idempotency key per run slot:

ti:{template_id}:{asset_id}:{engine}:{yyyy-mm-ddThh}

Give critical-tier tasks a higher priority so they start ahead of background ones in your queue. The receiver must verify Webhook-Signature on the raw body, reject stale timestamps, ignore "test": true, and answer 2xx within 15 seconds before handing work to your queue. Webhooks specifies all of this, and the AI visibility tracking guide has a complete receiver.

Storage schema

create table templates (
  template_id  text primary key,
  family       text not null,
  engine       text not null,       -- GOOGLE, GOOGLE_NEWS, CHATGPT
  pattern      text not null,       -- with {brand}, {domain}, {unique_string} placeholders
  severity     text not null,
  active       boolean not null default true
);

create table runs (
  task_id      uuid primary key,
  template_id  text not null references templates,
  asset_id     text not null,
  slot         timestamptz not null,
  status       text not null default 'QUEUED',
  result_count int,
  completed_at timestamptz,
  unique (template_id, asset_id, slot)
);

create table exposures (
  template_id  text not null,
  asset_id     text not null,
  url_key      text not null,       -- sha256 of normalized URL
  url          text not null,
  host         text not null,
  title        text,
  snippet      text,
  google_date  text,
  first_seen   timestamptz not null,
  last_seen    timestamptz not null,
  verdict      text,                -- null | confirmed | benign | owned | duplicate
  primary key (template_id, asset_id, url_key)
);

create table ai_observations (
  task_id      uuid primary key references runs,
  text         text not null,
  source_urls  text[] not null default '{}',
  mentioned_hosts text[] not null default '{}'
);

Deduplication and change detection

• Deliveries: only a non-final run is processed, so repeated deliveries are no-ops.
• URLs: normalize before hashing (lowercase scheme and host, drop fragments, strip tracking parameters and trailing slashes). Keep the original URL for the analyst.
• New vs seen: a new (template_id, asset_id, url_key) row is an exposure event; an existing row only updates last_seen.
• First-run suppression: the first run of a new template or asset marks everything seen without alerting.
• Owned hosts: results on your hosts are expected for some families (document exposure) and noise for others (impersonation), so classify per family.
• Silent templates: a template whose result_count stays 0 for many runs after returning results before is itself an alert for its owner.

End-to-end Python

Expand templates

import datetime as dt

HOOK = {"url": "https://hooks.example.com/ti"}
PRIORITY = {"critical": 10, "standard": 5, "background": 1}

def payload(engine, text):
    if engine == "CHATGPT":
        return {"prompt": text, "country": "US"}
    return {"query": text, "country": "US"}          # GOOGLE and GOOGLE_NEWS

def tier_tasks(db, tier, assets):
    slot = dt.datetime.now(dt.timezone.utc).replace(minute=0, second=0, microsecond=0)
    templates = db.execute("select template_id, engine, pattern from templates where active and severity = %s",
                           (tier,)).fetchall()
    tasks = []
    for template_id, engine, pattern in templates:
        for asset in assets:
            try:
                text = pattern.format(**asset["vars"])
            except KeyError:
                continue                                  # template needs a variable this asset lacks
            tasks.append({"taskType": engine, "payload": payload(engine, text), "priority": PRIORITY[tier],
                          "idempotencyKey": f"ti:{template_id}:{asset['id']}:{engine}:{slot:%Y-%m-%dT%H}",
                          "webhook": HOOK})
    return slot, tasks

asset["vars"] holds the placeholder values, for example {"brand": "acme", "domain": "acme.com"}. Submit in chunks of 500 and insert a runs row per accepted task.

Normalize and diff

apply runs on each verified, non-test delivery:

import hashlib, json, re
from urllib.parse import urlsplit, urlunsplit, parse_qsl, urlencode

OWNED = {"acme.com", "acme.io"}

def normalize(url):
    p = urlsplit(url)
    q = urlencode(sorted((k, v) for k, v in parse_qsl(p.query) if not k.lower().startswith(("utm_", "fbclid", "gclid"))))
    clean = urlunsplit((p.scheme.lower(), (p.hostname or "").lower(), p.path.rstrip("/") or "/", q, ""))
    return clean, hashlib.sha256(clean.encode()).hexdigest(), (p.hostname or "").lower()

def owned(host):
    return any(host == d or host.endswith("." + d) for d in OWNED)

def apply(db, d):
    task = d["task"]
    now = dt.datetime.now(dt.timezone.utc)
    with db.transaction():
        row = db.execute(
            """update runs r set status = %s, completed_at = %s from templates t
               where r.task_id = %s and r.status not in ('COMPLETED','FAILED') and t.template_id = r.template_id
               returning r.template_id, r.asset_id, t.engine, t.family""",
            (task["status"], now, task["id"])).fetchone()
        if not row or task["status"] != "COMPLETED":
            return
        template_id, asset_id, engine, family = row
        r = d["response"]["result"]
        first_run = not db.execute("select 1 from runs where template_id = %s and asset_id = %s "
                                   "and status = 'COMPLETED' and task_id <> %s limit 1",
                                   (template_id, asset_id, task["id"])).fetchone()
        if engine == "CHATGPT":
            hosts = sorted(set(re.findall(r"\b(?:[a-z0-9-]+\.)+[a-z]{2,}\b", r.get("text", "").lower())))
            db.execute("insert into ai_observations values (%s,%s,%s,%s) on conflict do nothing",
                       (task["id"], r.get("text", ""), [s["url"] for s in r.get("sources", [])], hosts))
            items = [{"link": s["url"], "title": s.get("label"), "snippet": s.get("description")}
                     for s in r.get("sources", [])]
        else:
            items = r.get("newsResults" if engine == "GOOGLE_NEWS" else "organicResults", [])
        db.execute("update runs set result_count = %s where task_id = %s", (len(items), task["id"]))
        for it in items:
            clean, key, host = normalize(it["link"])
            if family == "impersonation" and owned(host):
                continue
            new = db.execute(
                """insert into exposures values (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,null)
                   on conflict (template_id, asset_id, url_key) do update set last_seen = excluded.last_seen
                   returning (xmax = 0) as inserted""",
                (template_id, asset_id, key, clean, host, it.get("title"), it.get("snippet"), it.get("date"),
                 now, now)).fetchone()[0]
            if new and not first_run:
                emit_alert(template_id, asset_id, host, clean, it)

def emit_alert(template_id, asset_id, host, url, item):
    # Push to your SIEM or ticket queue. Do not fetch the URL from here.
    print(json.dumps({"template": template_id, "asset": asset_id, "host": host, "url": url,
                      "title": item.get("title"), "snippet": item.get("snippet")}))

xmax = 0 is a Postgres idiom that is true when the upsert inserted rather than updated. ChatGPT sources are stored as exposures for the social-chatter family, and domains mentioned in the answer text go to ai_observations for an analyst to compare with known infrastructure.

Metrics

-- New exposures and precision per family, last 30 days
select t.family,
       count(*)                                                          as new_exposures,
       avg((e.verdict = 'confirmed')::int) filter (where e.verdict is not null) as alert_precision,
       count(*) filter (where e.verdict is null)                         as untriaged
from exposures e join templates t using (template_id)
where e.first_seen > now() - interval '30 days'
group by t.family order by new_exposures desc;

-- Silent templates: active templates with no results in their last 5 completed runs
select template_id
from (select template_id, result_count,
             row_number() over (partition by template_id order by completed_at desc) as n
      from runs where status = 'COMPLETED') x
where n <= 5
group by template_id
having max(result_count) = 0 and count(*) = 5;

-- Watch-list coverage for the last daily cycle
select count(distinct asset_id) filter (where not exists (
         select 1 from runs r2 where r2.asset_id = r.asset_id and r2.slot > now() - interval '1 day'
           and r2.status <> 'COMPLETED'))::numeric
       / count(distinct asset_id) as coverage
from runs r where r.slot > now() - interval '1 day';

Cost planning

Price per request at the Starter tier: Google Search $0.00103 for one page ($0.00071 more per extra page), Google News $0.0002 (extra pages free), ChatGPT $0.0006. Failed tasks are not charged. A query with a search operator such as site:, inurl: or filetype: costs 5x on Google Search; every Google Search template above uses one. Higher tiers, reached by 30-day spend, cost less. See balance and pricing.

Worked example: a security team watches 6 brands and 25 domains.

That is about $33.12 over 30 days at Starter-tier prices. The multi-page template costs the most per run; check whether pages 2 and 3 ever produce confirmed exposures and cut them if not. Tiers are on pricing, and rate limits lists the queue limit that applies to a large watch list.

Pitfalls

Operator drift. Google’s handling of operators can change. Monitor silent templates and re-test templates monthly.

First-run floods. A new template returns everything already indexed. Suppress alerts on the first completed run per template and asset.

Your own infrastructure as noise. Owned hosts, CDNs and partners appear in many families. Maintain an owned-host list and classify per family, not globally.

Treating AI answers as intelligence. An assistant’s answer summarizing chatter can be wrong or out of date. Keep its sources and let analysts verify from them.

Fetching found URLs from the pipeline. Found links can be malicious. Keep the pipeline read-only on API data and let analysts open URLs in an isolated environment.

Relative and missing dates. date is display text when present and absent for many results. Use first seen for timing and treat date as a hint.

Scope. Only monitor assets you are responsible for or authorized to assess, and follow your organization’s rules for handling data you find.

Redirect wrappers. Some results carry redirectLink beside the destination link. Use link for identity.

Which engines fit

Request fields are on the Google Search engine page, and the quickstart wires a first query.

More use cases