Email Domain Reputation API: Detect Signup Fraud with WHOIS and DNS

https://whoisjson.com/api/v1

curl "https://whoisjson.com/api/v1/whois?domain=example.com" \
  -H "Authorization: TOKEN=YOUR_API_KEY"

curl "https://whoisjson.com/api/v1/nslookup?domain=example.com" \
  -H "Authorization: TOKEN=YOUR_API_KEY"

import requests

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://whoisjson.com/api/v1"
HEADERS = {"Authorization": f"TOKEN={API_KEY}"}


def email_domain(email: str) -> str:
    if "@" not in email:
        raise ValueError("Invalid email address")
    return email.rsplit("@", 1)[1].strip().lower()


def parse_remaining_requests(value: str | None) -> int | None:
    if value is None:
        return None
    try:
        return int(value)
    except ValueError:
        return None


def get_json(path: str, domain: str) -> tuple[dict, int | None]:
    response = requests.get(
        f"{BASE_URL}{path}",
        headers=HEADERS,
        params={"domain": domain},
        timeout=10,
    )
    response.raise_for_status()
    remaining = parse_remaining_requests(response.headers.get("Remaining-Requests"))
    return response.json(), remaining


def has_spf(txt_records: list) -> bool:
    return any(str(record).lower().startswith("v=spf1") for record in txt_records or [])


def score_email_domain(email: str) -> dict:
    domain = email_domain(email)
    whois, whois_remaining = get_json("/whois", domain)
    dns, dns_remaining = get_json("/nslookup", domain)

    score = 0
    reasons = []

    age = whois.get("age") or {}
    if age.get("isNewlyRegistered"):
        score += 35
        reasons.append("domain registered within 30 days")
    elif age.get("isYoung"):
        score += 15
        reasons.append("domain registered within 365 days")

    status_values = " ".join(whois.get("status") or []).lower()
    if any(flag in status_values for flag in ["clienthold", "serverhold", "pendingdelete"]):
        score += 20
        reasons.append("domain has hold or pending delete status")

    contacts = (whois.get("contacts") or {}).get("owner") or []
    if not contacts:
        score += 10
        reasons.append("owner contact is absent or redacted")

    mx_records = dns.get("MX") or []
    if not mx_records:
        score += 25
        reasons.append("no MX records")

    txt_records = dns.get("TXT") or []
    if not has_spf(txt_records):
        score += 10
        reasons.append("no SPF-like TXT record")

    dmarc_records = dns.get("DMARC") or []
    if not dmarc_records:
        score += 10
        reasons.append("no DMARC record")

    if score >= 75:
        decision = "manual_review_or_block"
    elif score >= 50:
        decision = "step_up_verification"
    elif score >= 25:
        decision = "allow_with_monitoring"
    else:
        decision = "allow"

    remaining_values = [
        value for value in [whois_remaining, dns_remaining]
        if value is not None
    ]

    return {
        "email": email,
        "domain": domain,
        "score": min(score, 100),
        "decision": decision,
        "reasons": reasons,
        "remaining_requests": min(remaining_values) if remaining_values else None,
    }


print(score_email_domain("[email protected]"))

const API_KEY = process.env.WHOISJSON_API_KEY;
const BASE_URL = 'https://whoisjson.com/api/v1';

function getEmailDomain(email) {
  const parts = String(email).toLowerCase().trim().split('@');
  if (parts.length !== 2 || !parts[1]) {
    throw new Error('Invalid email address');
  }
  return parts[1];
}

function parseRemainingRequests(value) {
  if (value === null) return null;
  const parsed = Number.parseInt(value, 10);
  return Number.isNaN(parsed) ? null : parsed;
}

async function fetchJson(path, domain) {
  const url = new URL(`${BASE_URL}${path}`);
  url.searchParams.set('domain', domain);

  const controller = new AbortController();
  const timeout = setTimeout(() => controller.abort(), 10000);

  try {
    const response = await fetch(url, {
      headers: { Authorization: `TOKEN=${API_KEY}` },
      signal: controller.signal
    });

    if (!response.ok) {
      return { error: `HTTP ${response.status}` };
    }

    return {
      data: await response.json(),
      remainingRequests: parseRemainingRequests(response.headers.get('Remaining-Requests'))
    };
  } finally {
    clearTimeout(timeout);
  }
}

function hasSpf(txtRecords = []) {
  return txtRecords.some((record) =>
    String(record).toLowerCase().startsWith('v=spf1')
  );
}

async function scoreSignupEmail(email) {
  const domain = getEmailDomain(email);
  const [whoisResult, dnsResult] = await Promise.all([
    fetchJson('/whois', domain),
    fetchJson('/nslookup', domain)
  ]);

  if (whoisResult.error || dnsResult.error) {
    return {
      email,
      domain,
      score: 30,
      decision: 'allow_with_monitoring',
      reasons: ['domain intelligence lookup incomplete'],
      remaining_requests: null
    };
  }

  const whois = whoisResult.data;
  const dns = dnsResult.data;
  let score = 0;
  const reasons = [];
  const age = whois.age || {};

  if (age.isNewlyRegistered) {
    score += 35;
    reasons.push('domain registered within 30 days');
  } else if (age.isYoung) {
    score += 15;
    reasons.push('domain registered within 365 days');
  }

  const status = (whois.status || []).join(' ').toLowerCase();
  if (status.includes('clienthold') || status.includes('serverhold') || status.includes('pendingdelete')) {
    score += 20;
    reasons.push('domain has hold or pending delete status');
  }

  const ownerContacts = (whois.contacts || {}).owner || [];
  if (!ownerContacts.length) {
    score += 10;
    reasons.push('owner contact is absent or redacted');
  }

  if (!(dns.MX || []).length) {
    score += 25;
    reasons.push('no MX records');
  }

  if (!hasSpf(dns.TXT || [])) {
    score += 10;
    reasons.push('no SPF-like TXT record');
  }

  if (!(dns.DMARC || []).length) {
    score += 10;
    reasons.push('no DMARC record');
  }

  const decision =
    score >= 75 ? 'manual_review_or_block' :
    score >= 50 ? 'step_up_verification' :
    score >= 25 ? 'allow_with_monitoring' :
    'allow';

  const remainingValues = [
    whoisResult.remainingRequests,
    dnsResult.remainingRequests
  ].filter((value) => value !== null);

  return {
    email,
    domain,
    score: Math.min(score, 100),
    decision,
    reasons,
    remaining_requests: remainingValues.length
      ? Math.min(...remainingValues)
      : null
  };
}

module.exports = { scoreSignupEmail };