Bot Detection for
Real Estate & PropTech

import requests

IPASIS_KEY = os.environ["IPASIS_API_KEY"]

def check_listing_access(request):
    """Gate listing detail/search endpoints"""
    ip = request.headers.get("X-Forwarded-For", request.remote_addr)

    resp = requests.get(
        "https://api.ipasis.com/v1/lookup",
        params={"ip": ip},
        headers={"Authorization": f"Bearer {IPASIS_KEY}"},
        timeout=3
    )
    data = resp.json()

    # Scraper detection logic
    is_scraper = (
        data.get("is_datacenter", False) or
        data.get("is_proxy", False) or
        data.get("risk_score", 0) >= 0.7
    )

    if is_scraper:
        # Option 1: Block entirely
        # return jsonify({"error": "Access denied"}), 403

        # Option 2: Return limited data (no photos, no price history)
        return serve_limited_listing(listing_id)

    # Option 3: Log for monitoring
    if data.get("risk_score", 0) >= 0.4:
        log_suspicious_access(ip, data)

    return serve_full_listing(listing_id)

const IPASIS_KEY = process.env.IPASIS_API_KEY;

async function validateLead(req, res) {
  const { name, email, phone, message } = req.body;
  const ip = req.headers["x-forwarded-for"]?.split(",")[0]
    || req.socket.remoteAddress;

  // Check IP risk
  const risk = await fetch(
    `https://api.ipasis.com/v1/lookup?ip=${ip}`,
    { headers: { Authorization: `Bearer ${IPASIS_KEY}` } }
  ).then(r => r.json());

  const riskScore = risk.risk_score || 0;
  const isVpn = risk.is_vpn || false;
  const isDatacenter = risk.is_datacenter || false;

  // Lead quality scoring
  if (riskScore >= 0.8 || risk.is_tor) {
    // Block — almost certainly a bot
    return res.status(403).json({
      error: "Unable to process your inquiry"
    });
  }

  if (riskScore >= 0.5 || isVpn || isDatacenter) {
    // Flag for manual review — don't auto-assign to agent
    await saveLead({
      name, email, phone, message,
      ip, riskScore, isVpn,
      status: "MANUAL_REVIEW",
      flagReason: `Risk: ${riskScore}, VPN: ${isVpn}`
    });
    return res.json({ status: "received" });
  }

  // Clean lead — route directly to agent
  await saveLead({
    name, email, phone, message,
    ip, riskScore,
    status: "NEW",
  });
  await notifyAgent(email, name, message);
  return res.json({ status: "received" });
}

def prescreen_application(request, application):
    """Pre-screen before running paid background check"""
    ip = request.headers.get("X-Forwarded-For", request.remote_addr)

    risk = requests.get(
        "https://api.ipasis.com/v1/lookup",
        params={"ip": ip},
        headers={"Authorization": f"Bearer {IPASIS_KEY}"},
        timeout=3
    ).json()

    flags = []
    risk_score = risk.get("risk_score", 0)

    # Geographic mismatch
    if risk.get("country_code") != "US":
        flags.append("non_us_ip")

    # Proxy/VPN detection
    if risk.get("is_residential_proxy"):
        flags.append("residential_proxy")
    if risk.get("is_vpn"):
        flags.append("vpn_detected")

    # Datacenter IP (bots don't rent apartments)
    if risk.get("is_datacenter"):
        flags.append("datacenter_ip")

    # High risk score
    if risk_score >= 0.6:
        flags.append(f"high_risk_{risk_score}")

    if flags:
        application.fraud_flags = flags
        application.status = "MANUAL_REVIEW"
        application.save()
        notify_property_manager(application, flags)
        return {"status": "under_review"}

    # Clean — proceed to paid background check
    application.status = "SCREENING"
    application.save()
    trigger_background_check(application)
    return {"status": "screening_initiated"}