Video KYC: How to Build It

from ipaddress import ip_address

# RBI requires the customer to be IN INDIA, the app to reject foreign IPs, and
# to detect IP spoofing. Check all of it BEFORE an officer joins the call --
# an officer's time is the most expensive thing in this pipeline.

def preflight(session):
    fail = []

    # 1. GEOTAG. Live GPS from the device, not an IP-derived guess.
    gps = session.get("gps")
    if not gps:
        fail.append("no_gps")                    # permission denied or spoofed
    elif not (6.0 <= gps["lat"] <= 37.5 and 68.0 <= gps["lon"] <= 97.5):
        fail.append("outside_india")             # rough India bounding box

    # 2. IP. Must resolve to India, and must not look like a tunnel.
    ip = session.get("ip")
    if not ip or session["ip_country"] != "IN":
        fail.append("foreign_ip")
    if session.get("is_vpn") or session.get("is_proxy") or session.get("is_hosting"):
        fail.append("ip_spoof_suspected")        # datacentre ASN on a phone = no

    # 3. DO THE TWO AGREE? GPS in Mumbai and IP in Frankfurt is the
    #    interesting case, and it is the one a spoofer produces.
    if "outside_india" not in fail and "foreign_ip" not in fail:
        if session.get("ip_distance_km", 0) > 500:
            fail.append("gps_ip_mismatch")

    # 4. Can the call actually work? A session that dies at step 6 wastes the
    #    officer AND the customer, and they rarely come back.
    if session.get("bandwidth_kbps", 0) < 400: fail.append("bandwidth_too_low")
    if not session.get("camera_ok"): fail.append("no_camera")
    if not session.get("mic_ok"): fail.append("no_mic")

    return {"proceed": not fail, "reasons": fail}

# WHAT TO CHECK
# [ ] GPS is captured live at the START and recorded INTO the video, not read
#     once and stored beside it. The recording itself must carry the coordinates
# [ ] tell the customer WHICH check failed, in plain words. "Please enable
#     location" gets you a session; "Verification failed" gets you a complaint
# [ ] a bounding box is a coarse first filter, not a border. Use it to reject
#     obvious cases fast, and a proper geocoder for the record
# [ ] datacentre and hosting ASNs on a consumer session are the strongest
#     spoofing signal you get. Treat hosting-provider IPs as disqualifying
# [ ] failures are LOGGED with the reason. The distribution tells you whether
#     you have a fraud problem or a UX problem, and they look identical in a
#     total
# [ ] never let a failed pre-flight silently fall back to a different KYC route.
#     That decision is a compliance decision, not an error handler

# Liveness has three generations, and most products are stuck in the first two.
#
#   2019: is this a PHOTO?          -- beaten by playing a video
#   2021: is this a VIDEO REPLAY?   -- beaten by a deepfake
#   2026: is this a DEEPFAKE, and is it even coming from the CAMERA?
#
# The 2026 question is INJECTION. An attacker does not hold a screen up to the
# lens -- they replace the camera feed with a virtual camera driver and send a
# synthetic face straight into your app. Every pixel-level liveness check in the
# world passes, because the pixels are perfect.

def assess_liveness(signals):
    checks = {
        # presentation attack: something held up to a real camera
        "pad_score":        signals["pad_score"] >= 0.90,
        # injection attack: the frames never came from a camera at all
        "camera_is_real":   signals["virtual_camera_detected"] is False,
        "driver_signed":    signals["camera_driver_trusted"],
        "frame_timing_ok":  signals["frame_interval_variance"] > 0.0001,
        # a synthetic stream is often TOO regular. Real cameras jitter.
        "sensor_noise_ok":  signals["sensor_noise_present"],
    }
    failed = [k for k, v in checks.items() if not v]
    if "camera_is_real" in failed or "driver_signed" in failed:
        return {"outcome": "reject_hard", "reason": "injection_suspected",
                "escalate": True}
    if failed:
        return {"outcome": "retry", "reason": failed, "attempts_allowed": 2}
    return {"outcome": "pass"}

# WHAT TO CHECK
# [ ] ASK EVERY VENDOR: "do you detect virtual-camera injection?" The answer
#     tells you whether you are buying a 2021 product or a 2026 one. It is the
#     single most useful question in this whole procurement
# [ ] ask for iBeta PAD certification and the LEVEL. Level 1 is photos and
#     screens; Level 2 is masks. Neither covers injection -- that is separate
# [ ] DO NOT require blinking or smiling. RBI's FAQ (Q20) says specific facial
#     gestures are not mandatory, and accommodation is required for customers
#     who cannot perform them. A gesture-gated flow excludes disabled customers
#     and is a conduct problem as well as an accessibility one
# [ ] an injection signal is a HARD reject and an escalation, not a retry. A
#     retry just tells the attacker which check fired
# [ ] log the raw scores, not the verdict. Thresholds change; the evidence of
#     what you saw should not have to be recomputed
# [ ] rehearse a failure. If your liveness vendor is down, what happens? The
#     answer must not be "approve anyway"

from datetime import datetime, timedelta, timezone
IST = timezone(timedelta(hours=5, minutes=30))

# THREE OUTCOMES. A two-outcome system either rejects good customers or
# approves bad sessions, and usually both.

def officer_decision(session, checks, officer):
    if checks["liveness"]["outcome"] == "reject_hard":
        return {"outcome": "reject", "reason": "injection_suspected",
                "reviewable": True}       # a person can still overturn it
    if not checks["documents_valid"] or checks["face_match"] < session["threshold"]:
        return {"outcome": "redo", "reason": "quality_or_match",
                "guidance": "explain WHICH part, in plain words"}
    return {"outcome": "recommend_approve", "by": officer["id"],
            "at": datetime.now(IST).isoformat()}

# STEP 8. The officer RECOMMENDS. A DIFFERENT team approves. The account does
# not exist until that second team has looked. This is a hard requirement and
# it is the step product teams discover three weeks before launch.

def concurrent_audit(session, auditor):
    assert auditor["team"] != session["officer_team"], "auditor must be independent"
    findings = []
    for need in ("recording_present", "gps_in_recording", "timestamps_present",
                 "consent_captured", "officer_credentials_recorded",
                 "questions_randomised", "aadhaar_redacted", "hash_matches"):
        if not session.get(need):
            findings.append(need)
    return {"activate": not findings, "findings": findings,
            "auditor": auditor["id"], "at": datetime.now(IST).isoformat()}

# WHAT TO CHECK
# [ ] the auditor is a DIFFERENT TEAM. Asserted in code, not in a policy
#     document. This is the most commonly faked control in the whole process
# [ ] account activation is gated on the audit result. If an account can exist
#     before the audit, the audit is decorative
# [ ] "redo" tells the customer which part failed. A blank retry produces the
#     same failure and then an abandoned application
# [ ] the hash from step 2 is re-verified HERE, at audit time. Checking it only
#     at write time proves nothing about the intervening days
# [ ] audit capacity is planned as a number. It is a person per N sessions and
#     it does not scale by deploying more servers -- this is the line that
#     silently caps your onboarding volume
# [ ] measure the audit FAILURE rate weekly. Near zero means the audit is a
#     rubber stamp; rising means something upstream broke
# [ ] rejections are reviewable by a human on request. "The system said no" is
#     not an answer you can give a customer about their bank account