Docs / Honeypot Lookup API / Dialer integration

Honeypot scrubbing for VICIdial, Asterisk and FreeSWITCH

Three ways to use the lookup, in the order you should reach for them. Scrubbing a list before you load it is cheapest and safest. Scrubbing leads already loaded catches numbers that became known after you imported them. Checking at dial time is the last resort, because it puts a network call in front of every dial.

Which approach to use

1. Scrub a list before you load it

Feed the API up to 100 numbers per request and write out two files: the numbers that are safe to load, and the ones that are known honeypots. Run this on the CSV before it ever touches VICIdial.

#!/usr/bin/env python3
"""Scrub a lead file against the AMDY honeypot index.

  ./scrub.py leads.csv          -> leads.clean.csv + leads.honeypot.csv

Reads the phone number from the first column. Adjust PHONE_COL if yours differs.
"""
import csv, sys, json, urllib.request

KEY      = open("/etc/amdy/hp-key").read().strip()
ENDPOINT = "https://app.amdy.io/hp/v1/honeypot/batch"
BATCH    = 100
PHONE_COL = 0

def check(numbers):
    """Return the set of numbers the index knows as honeypots."""
    req = urllib.request.Request(
        ENDPOINT,
        data=json.dumps({"phones": numbers}).encode(),
        headers={"X-API-Key": KEY, "Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=30) as r:
        body = json.load(r)
    return {row["phone"] for row in body["results"] if row["honeypot"]}

src = sys.argv[1]
rows = list(csv.reader(open(src, newline="")))
phones = [r[PHONE_COL] for r in rows if r and r[PHONE_COL].strip()]

bad = set()
for i in range(0, len(phones), BATCH):
    bad |= check(phones[i:i + BATCH])

def last10(p):
    d = "".join(c for c in p if c.isdigit())
    return d[1:] if len(d) == 11 and d.startswith("1") else d

clean = csv.writer(open(src.replace(".csv", ".clean.csv"), "w", newline=""))
dirty = csv.writer(open(src.replace(".csv", ".honeypot.csv"), "w", newline=""))
kept = dropped = 0
for r in rows:
    if r and last10(r[PHONE_COL]) in bad:
        dirty.writerow(r); dropped += 1
    else:
        clean.writerow(r); kept += 1

print(f"{kept} clean, {dropped} honeypots removed ({dropped / max(len(rows),1):.1%})")

2. Scrub leads already in VICIdial

Numbers become known honeypots after you loaded them, so a list that was clean in June may not be clean today. This script walks the leads in a list, checks them in batches, and parks the hits on a status your campaign does not dial.

First create the status once, in Admin → Statuses: status HPOT, description "Honeypot", and leave selectable unchecked. A lead whose status is not in the campaign's dial statuses is never pulled into the hopper, so this takes it out of rotation without deleting anything.

#!/usr/bin/env python3
"""Nightly honeypot scrub of VICIdial leads.  cron: 15 2 * * *

Marks known honeypots as status HPOT so the hopper stops dialing them.
Only looks at leads that are currently dialable, and only at lists you name.
"""
import json, urllib.request, pymysql

KEY   = open("/etc/amdy/hp-key").read().strip()
LISTS = [101, 102]            # list_id values to scrub
BATCH = 100

db = pymysql.connect(host="localhost", user="cron", password="1234",
                     database="asterisk", autocommit=True)

def honeypots(numbers):
    req = urllib.request.Request(
        "https://app.amdy.io/hp/v1/honeypot/batch",
        data=json.dumps({"phones": numbers}).encode(),
        headers={"X-API-Key": KEY, "Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=30) as r:
        return {x["phone"] for x in json.load(r)["results"] if x["honeypot"]}

with db.cursor() as cur:
    cur.execute(
        """SELECT DISTINCT phone_number FROM vicidial_list
            WHERE list_id IN %s AND status NOT IN ('HPOT','DNC','DNCL')""",
        (tuple(LISTS),),
    )
    phones = [r[0] for r in cur.fetchall()]

flagged = 0
for i in range(0, len(phones), BATCH):
    bad = honeypots(phones[i:i + BATCH])
    if not bad:
        continue
    with db.cursor() as cur:
        cur.execute(
            """UPDATE vicidial_list SET status = 'HPOT'
                WHERE list_id IN %s AND RIGHT(phone_number, 10) IN %s""",
            (tuple(LISTS), tuple(bad)),
        )
        flagged += cur.rowcount

print(f"checked {len(phones)} numbers, flagged {flagged} leads as HPOT")

If you would rather these numbers never be dialable on any campaign, add them to your internal DNC list in Admin → Lists → DNC instead of, or as well as, setting the status. Confirm the DNC table layout for your VICIdial version before writing to it directly; setting the lead status is the version-safe option.

3. Check at dial time (Asterisk AGI)

Only do this if you cannot scrub ahead of time. It adds a round trip in front of every dial (normally well under 100 ms). The script fails open: if the lookup is slow, unreachable or the key is wrong, the call goes ahead, and it never holds a call longer than 2 seconds. It runs on Python 3.4 and newer (ViciBox 7 onwards) with no extra modules.

Install it on the dialer, as root, with your Honeypot API key:

curl -fsSL https://download.amdy.io/hp-check.agi -o /var/lib/asterisk/agi-bin/hp-check.agi
chmod 755 /var/lib/asterisk/agi-bin/hp-check.agi
mkdir -p /etc/amdy && echo 'YOUR_HONEYPOT_API_KEY' > /etc/amdy/hp-key && chmod 600 /etc/amdy/hp-key

It sets HONEYPOT to YES, NO or UNKNOWN, plus HONEYPOT_PHONE (the 10-digit number checked), HONEYPOT_DETECTIONS and HONEYPOT_REASON (why it is UNKNOWN: no_key, not_us_number, timeout, http_401, …).

Pass the number in any format: 2125551234, 12125551234, +1 (212) 555-1234, or with a dial prefix in front of the 1 (912125551234). It is checked as the same 10 digits AMDY stores. A prefix without the 1 (92125551234) is not guessed; strip it with ${EXTEN:1}.

ViciDial: add two lines before the Dial() of your carrier's dialplan entry, and a hangup label:

exten => _91NXXNXXXXXX,1,AGI(agi://127.0.0.1:4577/call_log)
exten => _91NXXNXXXXXX,n,AGI(hp-check.agi,${EXTEN})
exten => _91NXXNXXXXXX,n,GotoIf($["${HONEYPOT}" = "YES"]?hp_block)
exten => _91NXXNXXXXXX,n,Dial(${TESTSIPTRUNK}/${EXTEN:1},,tTo)
exten => _91NXXNXXXXXX,n,Hangup()
exten => _91NXXNXXXXXX,n(hp_block),NoOp(honeypot suppressed: ${HONEYPOT_PHONE})
exten => _91NXXNXXXXXX,n,Hangup(21)

Any other Asterisk dialer:

exten => _1NXXNXXXXXX,1,AGI(hp-check.agi,${EXTEN})
 same => n,GotoIf($["${HONEYPOT}" = "YES"]?blocked)
 same => n,Dial(SIP/trunk/${EXTEN},55,o)
 same => n,Hangup()
 same => n(blocked),NoOp(honeypot suppressed: ${HONEYPOT_PHONE})
 same => n,Hangup(21)

Optional settings go in /etc/amdy/hp.conf: HP_TIMEOUT=2, and HP_TLS_VERIFY=0 only on an old box whose certificate bundle cannot verify the API.

FreeSWITCH

Same idea with the curl application. Set a short timeout and treat any non-answer as unknown.

<extension name="honeypot_check">
  <condition field="destination_number" expression="^(\d{10,11})$">
    <action application="set" data="curl_timeout=2"/>
    <action application="curl"
            data="https://app.amdy.io/hp/v1/honeypot?phone=$1 get \
                  header X-API-Key: ${amdy_hp_key}"/>
    <action application="set" data="hp=${curl_response_data}"/>
    <action application="log" data="INFO honeypot lookup: ${hp}"/>
    <!-- suppress only on an explicit true; anything else dials -->
    <action application="hangup" data="CALL_REJECTED"
            inline="true" nested="true"
            expression="${hp}" pattern="\"honeypot\":true"/>
    <action application="bridge" data="sofia/gateway/trunk/$1"/>
  </condition>
</extension>

Any other dialer

It is one GET with a header, so anything that can make an HTTP call can use it. Node, for a pre-dial hook or a CRM plugin:

const res = await fetch(
  `https://app.amdy.io/hp/v1/honeypot?phone=${phone}`,
  { headers: { 'X-API-Key': process.env.AMDY_HP_KEY }, signal: AbortSignal.timeout(2000) }
);

if (res.status === 403) throw new Error('honeypot lookup not subscribed');
if (!res.ok) return { honeypot: false, known: false };   // fail open

const { honeypot, detections } = await res.json();
if (honeypot) console.warn(`suppressing ${phone} — seen ${detections} times`);

Getting it right in production

• Act on true, not on false. A true is high confidence: suppress it. A false means the number is not known to us, which is not the same as clean.
• Re-scrub on a schedule. Coverage grows daily. Cache a true answer as long as you like, but do not cache a false forever. Weekly is a reasonable cadence for an active list.
• Fail open. If the lookup errors or times out, dial the call. A scrubbing service should never become the reason your floor goes quiet.
• Batch whenever you can. One request of 100 numbers beats 100 requests: it counts once against your requests-per-second limit.
• Watch for 403. It means the key is fine but the subscription is not active. Alert on it: a silently lapsed subscription looks exactly like a clean list.
• Keep what you removed. Write suppressed numbers to their own file or status rather than deleting the lead, so you can audit a suppression later.

Honeypot Lookup API reference · All docs