Carrier Edge에서 모든 인바운드 호출을 스크리닝: 178줄의 Python에서 사기 방화벽

작성자

카테고리:

← 피드로
DEV Community · Harpreet Singh Seehra · 2026-07-23 개발(SW)

Harpreet Singh Seehra

Originally published on lowlatencyclub.ai

Screen Every Inbound Call at the Carrier Edge: A Fraud Firewall in 178 Lines of Python

Every inbound call is a decision. Is this caller legitimate, or is it a scammer, robocaller, or toll-fraud bot? Most systems make that decision too late — after the call reaches the application layer, consuming resources and potentially connecting to a human agent who wastes 5 minutes on a fake call.

The Edge Fraud Firewall flips the order. It screens every inbound call at the carrier edge — before your app sees it — using three Telnyx APIs on a single platform: Number Lookup for caller identity, AI Inference for risk classification, and Call Control for reject/forward/route decisions.

The entire firewall is 178 lines of Python in a single Flask file.

How It Works

When a call hits a Telnyx number, the platform sends a call.initiated webhook to the Flask server. Before doing anything else, the server verifies the Ed25519 webhook signature — if the request is not from Telnyx, it returns 401 and stops.

For verified webhooks, the screening pipeline runs in order:

  1. Blocklist check — if the caller’s number is in the in-memory blocklist, reject immediately. No lookup, no AI call.
  2. Number Lookup — returns carrier name, line type (landline/voip/mobile), and country code.
  3. AI classification — the lookup data goes to an OpenAI-compatible chat completions endpoint with a system prompt that returns exactly one word: CLEAN, SUSPICIOUS, or BLOCK.
  4. Route decision — based on the classification:
    • CLEAN → answer and forward to your real number
    • SUSPICIOUS → answer and route to a honeypot (endless hold loop)
    • BLOCK → reject and add to the blocklist

The AI Classifier

The system prompt is deliberately constrained — one word only:

def classify_caller(phone, lookup_data):
    try:
        resp = requests.post(INFERENCE_URL, headers=HEADERS, timeout=15, json={
            "model": AI_MODEL,
            "messages": [
                {"role": "system", "content": "You are a fraud detection system. Analyze caller data and respond with ONLY one word: CLEAN, SUSPICIOUS, or BLOCK."},
                {"role": "user", "content": f"Caller: {phone}\nCarrier: {lookup_data.get('carrier', {}).get('name', 'unknown')}\nType: {lookup_data.get('carrier', {}).get('type', 'unknown')}\nCountry: {lookup_data.get('country_code', 'unknown')}"}
            ]
        })
        if resp.ok:
            return resp.json()["choices"][0]["message"]["content"].strip().upper()
    except Exception as e:
        app.logger.error("Classification failed: %s", e)
    return "CLEAN"

Enter fullscreen mode Exit fullscreen mode

One-word output keeps inference latency to a single token and makes parsing trivial — no JSON, no regex, just a string comparison.

If the AI call fails for any reason, the function defaults to CLEAN. This is a safe-fail — if the screening service is unavailable, legitimate callers still get through.

The Honeypot

When the AI classifies a call as SUSPICIOUS, the firewall routes it to a honeypot instead of rejecting or forwarding. The honeypot answers the call and plays a hold message, then loops indefinitely:

# call.answered → honeypot flow
if flow == "honeypot":
    requests.post(f".../calls/{call_control_id}/actions/speak",
                  json={"payload": "Please hold while I connect you to a specialist.",
                        "voice": "female", "language": "en-US",
                        "client_state": encode_state({"flow": "honeypot_hold"})})

# call.speak.ended → loop forever
if state.get("flow") == "honeypot_hold":
    time.sleep(2)
    requests.post(f".../calls/{call_control_id}/actions/speak",
                  json={"payload": "All specialists are currently busy. Please continue to hold.",
                        "client_state": encode_state({"flow": "honeypot_hold"})})

Enter fullscreen mode Exit fullscreen mode

The scammer stays on the line, burning their time and resources.

State Management with client_state

Telnyx Call Control is webhook-driven — each call action triggers a new webhook event. To track which flow a call is in (forward vs honeypot), the app uses client_state, a base64-encoded JSON blob that Telnyx passes back in the next webhook:

def encode_state(data):
    return base64.b64encode(json.dumps(data).encode()).decode()

def decode_state(b64):
    try:
        return json.loads(base64.b64decode(b64).decode())
    except Exception:
        return {}

Enter fullscreen mode Exit fullscreen mode

No external database needed — the state travels with the webhook.

Webhook Signature Verification

The first thing the webhook handler does is verify the request is from Telnyx:

try:
    client.webhooks.unwrap(request.get_data(as_text=True), headers=dict(request.headers))
except Exception:
    return jsonify({"error": "invalid signature"}), 401

Enter fullscreen mode Exit fullscreen mode

The Telnyx Python SDK verifies the Ed25519 signature against the raw request body. If verification fails, the request is rejected before any processing happens.

Setup

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/edge-fraud-firewall-python
cp .env.example .env
pip install -r requirements.txt
python app.py
ngrok http 5000

Enter fullscreen mode Exit fullscreen mode

Configure the webhook URL in the Telnyx Portal — set the Call Control Application webhook URL to https://<your-ngrok-id>.ngrok.io/webhooks/voice.

Why Screen at the Carrier Edge?

Fraud screening typically runs at the application layer — after the call is already connected. By that point, the fraud has already consumed SIP trunks, media servers, and agent time.

Screening at the carrier edge means the decision happens on the same network that carries the call. Number Lookup and AI Inference run on Telnyx infrastructure, not a third-party API called over the public internet. The screen-then-route decision is fast enough to sit inline on every inbound call.

Going to Production

This sample uses in-memory state. For production:

  • Replace the in-memory blocklist with Redis
  • Add rate limiting to the AI classification call
  • Set up monitoring on classification latency and block rate
  • Use gunicorn -w 4 app:app as a process manager
  • Configure failover webhook URLs in the Telnyx Portal

Resources

Related Examples

원문에서 계속 ↗

코멘트

답글 남기기

이메일 주소는 공개되지 않습니다. 필수 필드는 *로 표시됩니다