Build your own Account Connect endpoints

Customer ──► Bot: "What's my balance?"
Bot      ──► asks for full name + postcode (or email / mobile)
Bot      ──► POST {base}/verify-identity  {full_name, postcode, email, mobile_number}
               YOUR SYSTEM finds the account and TEXTS A ONE-TIME CODE to the mobile
               number already on the account (never to a number the customer types)
             ◄── {"success": true, "otp_token": "…"}
Customer ──► types the code into the chat
Bot      ──► POST {base}/verify-otp       {otp_token, otp_code}
             ◄── {"success": true, "session_token": "…"}
Bot      ──► POST {base}/account-summary  {session_token}  ◄── {"success": true, "account": {…}}
Bot      ──► POST {base}/orders           {session_token}  ◄── {"success": true, "orders": […]}
Bot      ──► POST {base}/invoices         {session_token}  ◄── {"success": true, "invoices": […]}
Bot      ──► answers from that data only

{"success": false, "message": "We could not match those details to an account."}

{
  "success": true,
  "orders": [
    {"order_number": "ORD-50231", "placed": "2026-09-20", "status": "dispatched",
     "courier": "Royal Mail Tracked 48", "tracking": "TT123456789GB"}
  ]
}

{
  "success": true,
  "invoices": [
    {"invoice": "INV-7781", "date": "2026-09-01", "amount": "£19.95", "status": "paid"}
  ]
}

{"success": false, "message": "Session expired — please verify again."}

"""Account Connect endpoints for Silux Chat — reference implementation (Python 3.9+, Flask).

    pip install flask
    ACCOUNT_CONNECT_API_KEY=change-me flask --app account_connect run --port 8080

Run it behind your HTTPS reverse proxy. Replace the functions marked YOUR SYSTEM with
lookups in your own database, and send_sms() with your SMS provider.
"""
import hashlib
import hmac
import os
import secrets
import threading
import time
from typing import Optional

from flask import Flask, jsonify, request

API_KEY = os.environ["ACCOUNT_CONNECT_API_KEY"]      # the key you enter on the dashboard
OTP_TTL, SESSION_TTL, MAX_ATTEMPTS = 600, 1800, 5    # 10 minutes, 30 minutes, 5 wrong codes

app = Flask(__name__)
_lock = threading.Lock()
_otps = {}       # otp_token -> {"customer_id", "code_hash", "exp", "tries"}  (use Redis/DB in production)
_sessions = {}   # session_token -> {"customer_id", "exp"}

TEST_CUSTOMER = {"id": "TEST", "full_name": "Test Customer", "postcode": "TE5 7ST",
                 "email": "test@example.com", "mobile": "07700900000"}


# ---------------- YOUR SYSTEM: replace these ----------------
def find_customer(full_name, postcode, email, mobile) -> Optional[dict]:
    """Return {"id", "mobile"} when the full name AND one of postcode/email/mobile match."""
    def n(s):
        return "".join((s or "").lower().split())
    c = TEST_CUSTOMER
    if n(full_name) == n(c["full_name"]) and (
            (postcode and n(postcode) == n(c["postcode"])) or (email and n(email) == n(c["email"]))
            or (mobile and n(mobile) == n(c["mobile"]))):
        return c
    return None   # <- look the customer up in YOUR database here


def account_for(customer_id: str) -> dict:
    return {"account_number": customer_id, "name": "Test Customer", "status": "active",
            "plan": "Sample plan", "balance_due": "£0.00"}


def orders_for(customer_id: str) -> list:
    return [{"order_number": "ORD-1", "placed": "2026-09-20", "status": "dispatched",
             "courier": "Royal Mail Tracked 48", "tracking": "TT000000000GB"}]


def invoices_for(customer_id: str) -> list:
    return [{"invoice": "INV-1", "date": "2026-09-01", "amount": "£19.95", "status": "paid"}]


def send_sms(mobile: str, text: str) -> None:
    pass   # <- call YOUR SMS provider here
# ------------------------------------------------------------


def _hash(code: str) -> str:
    return hashlib.sha256(code.encode()).hexdigest()


def _fail(status: int, message: str):
    return jsonify({"success": False, "message": message}), status


@app.before_request
def _check_key():
    key = request.headers.get("X-API-Key", "")
    if not key or not hmac.compare_digest(key.encode(), API_KEY.encode()):
        return _fail(401, "Missing or wrong X-API-Key")
    return None


def _body() -> dict:
    data = request.get_json(silent=True)
    return data if isinstance(data, dict) else {}


def _s(d: dict, k: str) -> str:
    return str(d.get(k) or "").strip()


@app.post("/verify-identity")
def verify_identity():
    b = _body()
    c = find_customer(_s(b, "full_name"), _s(b, "postcode"), _s(b, "email"), _s(b, "mobile_number"))
    if not c:   # never say WHICH detail was wrong
        return jsonify({"success": False, "message": "We could not match those details to an account."})
    code = "123456" if c["id"] == "TEST" else f"{secrets.randbelow(10**6):06d}"
    token = secrets.token_urlsafe(24)
    with _lock:
        _otps[token] = {"customer_id": c["id"], "code_hash": _hash(code), "exp": time.time() + OTP_TTL, "tries": 0}
    if c["id"] != "TEST":   # the test customer is never texted
        send_sms(c["mobile"], f"Your verification code is {code}. It expires in 10 minutes.")
    return jsonify({"success": True, "otp_token": token, "message": f"Code sent to mobile ending {c['mobile'][-3:]}"})


@app.post("/verify-otp")
def verify_otp():
    b = _body()
    token = _s(b, "otp_token")
    with _lock:
        o = _otps.get(token)
        if not o or o["exp"] < time.time():
            _otps.pop(token, None)
            return jsonify({"success": False, "message": "That code has expired — please ask for a new one."})
        if not hmac.compare_digest(o["code_hash"], _hash(_s(b, "otp_code"))):
            o["tries"] += 1
            if o["tries"] >= MAX_ATTEMPTS:
                _otps.pop(token, None)
                return jsonify({"success": False, "message": "Too many wrong codes — please ask for a new one."})
            return jsonify({"success": False, "message": "That code is not correct."})
        _otps.pop(token, None)   # a code works once
        session = secrets.token_urlsafe(32)
        _sessions[session] = {"customer_id": o["customer_id"], "exp": time.time() + SESSION_TTL}
    return jsonify({"success": True, "session_token": session})


def _customer_id() -> Optional[str]:
    with _lock:
        s = _sessions.get(_s(_body(), "session_token"))
    return s["customer_id"] if s and s["exp"] >= time.time() else None


@app.post("/account-summary")
def account_summary():
    cid = _customer_id()
    return jsonify({"success": True, "account": account_for(cid)}) if cid else _fail(401, "Session expired — please verify again.")


@app.post("/orders")
def orders():
    cid = _customer_id()
    return jsonify({"success": True, "orders": orders_for(cid)}) if cid else _fail(401, "Session expired — please verify again.")


@app.post("/invoices")
def invoices():
    cid = _customer_id()
    return jsonify({"success": True, "invoices": invoices_for(cid)}) if cid else _fail(401, "Session expired — please verify again.")

# .htaccess in the same folder as account-connect.php (Apache)
RewriteEngine On
RewriteRule ^(verify-identity|verify-otp|account-summary|orders|invoices)$ account-connect.php?action=$1 [L,QSA]

# Never serve the store file
<Files ".account-connect-store.json">
  Require all denied
</Files>

BASE=http://localhost:8080          # PHP: http://localhost:8080/account-connect.php
KEY=change-me                       # the key your implementation expects

# 1. Wrong key -> HTTP 401
curl -s -w " [%{http_code}]\n" -X POST $BASE/verify-identity -H "X-API-Key: wrong" \
  -H "Content-Type: application/json" -d '{}'

# 2. Wrong details -> 200 with success:false (and no hint about WHICH detail)
curl -s -X POST $BASE/verify-identity -H "X-API-Key: $KEY" -H "Content-Type: application/json" \
  -d '{"full_name":"Test Customer","postcode":"XX1 1XX"}'

# 3. The test customer -> otp_token
OTP=$(curl -s -X POST $BASE/verify-identity -H "X-API-Key: $KEY" -H "Content-Type: application/json" \
  -d '{"full_name":"Test Customer","postcode":"TE5 7ST"}' \
  | python3 -c 'import sys,json;print(json.load(sys.stdin)["otp_token"])')

# 4. Code 123456 -> session_token
SES=$(curl -s -X POST $BASE/verify-otp -H "X-API-Key: $KEY" -H "Content-Type: application/json" \
  -d "{\"otp_token\":\"$OTP\",\"otp_code\":\"123456\"}" \
  | python3 -c 'import sys,json;print(json.load(sys.stdin)["session_token"])')

# 5. The three account endpoints
for EP in account-summary orders invoices; do
  curl -s -X POST $BASE/$EP -H "X-API-Key: $KEY" -H "Content-Type: application/json" \
    -d "{\"session_token\":\"$SES\"}"; echo
done

# 6. An unknown session -> HTTP 401
curl -s -w " [%{http_code}]\n" -X POST $BASE/orders -H "X-API-Key: $KEY" \
  -H "Content-Type: application/json" -d '{"session_token":"made-up"}'

{"success":false,"message":"Missing or wrong X-API-Key"} [401]
{"success":false,"message":"We could not match those details to an account."}
{"success":true,"account":{"account_number":"TEST","name":"Test Customer","status":"active","plan":"Sample plan","balance_due":"£0.00"}}
{"success":true,"orders":[{"order_number":"ORD-1","placed":"2026-09-20","status":"dispatched","courier":"Royal Mail Tracked 48","tracking":"TT000000000GB"}]}
{"success":true,"invoices":[{"invoice":"INV-1","date":"2026-09-01","amount":"£19.95","status":"paid"}]}
{"success":false,"message":"Session expired — please verify again."} [401]

wrong code 1-4  -> {"success":false,"message":"That code is not correct."}
wrong code 5    -> {"success":false,"message":"Too many wrong codes — please ask for a new one."}
any further try -> {"success":false,"message":"That code has expired — please ask for a new one."}