441 lines
21 KiB
Python
441 lines
21 KiB
Python
#!/usr/bin/env python3
|
||
"""shkor.com backend (stdlib only): email capture + lab applications with KYC/KYB via Didit.
|
||
|
||
POST /api/subscribe form email -> subs
|
||
POST /api/lab/apply JSON -> portal apply pipeline (42.cil.by /api/studio/apply) + Didit sessions
|
||
GET /api/lab/status?t= application + verification status (refreshed from Didit)
|
||
POST /api/autopilot/start JSON name/email -> Didit KYC session (Autopilot $99, shkor.com/start)
|
||
GET /api/autopilot/status?t= stage: kyc -> pay -> deploying -> ready (panel url + login)
|
||
POST /api/autopilot/checkout JSON t -> Stripe Checkout via portal onboarding (only after KYC Approved)
|
||
|
||
Env:
|
||
SHKOR_DB sqlite path (default /var/lib/shkor-subscribe/subs.db)
|
||
PORT default 8097
|
||
PUBLIC_BASE default https://shkor.com (used for Didit callback)
|
||
PORTAL_URL default http://127.0.0.1:8899
|
||
DIDIT_ENV_FILE file with DIDIT_API_KEY= (default /root/coop/.env)
|
||
DIDIT_KYC_WORKFLOW_ID default: Free KYC workflow
|
||
DIDIT_KYB_WORKFLOW_ID empty -> KYB disabled (company details are still collected)
|
||
TG_ENV_FILE file with APPLY42_TOKEN= (default /root/coop/social/.env)
|
||
TG_CHAT_ID default 7569223131
|
||
"""
|
||
import json, os, re, secrets, sqlite3, threading, time, urllib.error, urllib.request
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from urllib.parse import parse_qs, urlsplit
|
||
|
||
DB = os.environ.get("SHKOR_DB") or os.environ.get("SUBS_DB", "/var/lib/shkor-subscribe/subs.db")
|
||
PORT = int(os.environ.get("PORT", "8097"))
|
||
PUBLIC_BASE = os.environ.get("PUBLIC_BASE", "https://shkor.com").rstrip("/")
|
||
PORTAL_URL = os.environ.get("PORTAL_URL", "http://127.0.0.1:8899").rstrip("/")
|
||
DIDIT = "https://verification.didit.me"
|
||
KYC_WF = os.environ.get("DIDIT_KYC_WORKFLOW_ID", "4efa0ea4-3ddd-4810-8aa7-5f3fbf969ff9")
|
||
KYB_WF = os.environ.get("DIDIT_KYB_WORKFLOW_ID", "")
|
||
TG_CHAT = os.environ.get("TG_CHAT_ID", "7569223131")
|
||
FINAL = {"Approved", "Declined", "Abandoned", "Expired"}
|
||
EMAIL_RE = re.compile(r"^[^@\s]{1,64}@[^@\s]+\.[A-Za-z]{2,}$")
|
||
RATE = {} # (bucket, ip) -> [timestamps]
|
||
# портал провижнит на первом опросе после оплаты — не опрашивать параллельно (двойной тенант)
|
||
AP_LOCK = threading.Lock()
|
||
|
||
|
||
def env_value(path, key):
|
||
try:
|
||
for line in open(path):
|
||
k, _, v = line.strip().partition("=")
|
||
if k.strip() == key:
|
||
return v.strip()
|
||
except OSError:
|
||
pass
|
||
return ""
|
||
|
||
|
||
def didit_key():
|
||
return env_value(os.environ.get("DIDIT_ENV_FILE", "/root/coop/.env"), "DIDIT_API_KEY")
|
||
|
||
|
||
def db():
|
||
c = sqlite3.connect(DB, timeout=10)
|
||
c.row_factory = sqlite3.Row
|
||
c.executescript("""
|
||
create table if not exists subs (
|
||
email text primary key, source text, ip text, ua text,
|
||
created_at text default (datetime('now')));
|
||
create table if not exists lab_apps (
|
||
token text primary key, app_id integer, portal_track text,
|
||
name text, email text, project_name text,
|
||
company_name text, company_country text, company_reg text,
|
||
ip text, created_at text default (datetime('now')));
|
||
create table if not exists ap_orders (
|
||
token text primary key, name text, email text, company text, ip text,
|
||
kyc_session text, kyc_url text, kyc_status text, kyc_notified text default '',
|
||
pending_id text default '', paid_notified integer default 0,
|
||
created_at text default (datetime('now')), updated_at text default (datetime('now')));
|
||
create table if not exists lab_checks (
|
||
token text, kind text, session_id text, url text, status text,
|
||
notified text default '', updated_at text default (datetime('now')),
|
||
primary key (token, kind));
|
||
""")
|
||
return c
|
||
|
||
|
||
def limited(bucket, ip, n, window=3600):
|
||
now = time.time()
|
||
hits = [t for t in RATE.get((bucket, ip), []) if now - t < window]
|
||
if len(hits) >= n:
|
||
return True
|
||
RATE[(bucket, ip)] = hits + [now]
|
||
return False
|
||
|
||
|
||
def http_json(url, data=None, headers=None, method=None, timeout=20):
|
||
body = json.dumps(data).encode() if data is not None else None
|
||
h = {"Content-Type": "application/json", **(headers or {})}
|
||
req = urllib.request.Request(url, data=body, headers=h, method=method or ("POST" if body else "GET"))
|
||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||
return json.loads(r.read().decode())
|
||
|
||
|
||
def didit_session(workflow, vendor, callback):
|
||
return http_json(f"{DIDIT}/v3/session/", {"workflow_id": workflow, "vendor_data": vendor, "callback": callback},
|
||
{"x-api-key": didit_key()})
|
||
|
||
|
||
def didit_status(session_id):
|
||
return http_json(f"{DIDIT}/v3/session/{session_id}/decision/", headers={"x-api-key": didit_key()}).get("status")
|
||
|
||
|
||
def tg(text):
|
||
tok = env_value(os.environ.get("TG_ENV_FILE", "/root/coop/social/.env"), "APPLY42_TOKEN")
|
||
if not tok:
|
||
return
|
||
try:
|
||
http_json(f"https://api.telegram.org/bot{tok}/sendMessage", {"chat_id": TG_CHAT, "text": text}, timeout=10)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def refresh(c, token, force=False):
|
||
"""Pull fresh Didit statuses for unfinished checks; notify admin once per final status."""
|
||
app = c.execute("select * from lab_apps where token=?", (token,)).fetchone()
|
||
for ch in c.execute("select * from lab_checks where token=?", (token,)).fetchall():
|
||
if not ch["session_id"] or ch["status"] in FINAL:
|
||
continue
|
||
try:
|
||
st = didit_status(ch["session_id"]) or ch["status"]
|
||
except Exception:
|
||
continue
|
||
c.execute("update lab_checks set status=?, updated_at=datetime('now') where token=? and kind=?",
|
||
(st, token, ch["kind"]))
|
||
if st in FINAL and ch["notified"] != st:
|
||
icon = "✅" if st == "Approved" else "⛔"
|
||
tg(f"{icon} {ch['kind'].upper()} {st} — заявка в лабу #{app['app_id']}\n"
|
||
f"👤 {app['name']} · {app['email']}\n🚀 {app['project_name']}"
|
||
+ (f"\n🏢 {app['company_name']}" if ch["kind"] == "kyb" else ""))
|
||
c.execute("update lab_checks set notified=? where token=? and kind=?", (st, token, ch["kind"]))
|
||
c.commit()
|
||
|
||
|
||
def poller():
|
||
while True:
|
||
time.sleep(300)
|
||
try:
|
||
c = db()
|
||
toks = [r[0] for r in c.execute(
|
||
"select distinct token from lab_checks where status not in ('Approved','Declined','Abandoned','Expired') "
|
||
"and updated_at > datetime('now','-30 days')")]
|
||
for t in toks:
|
||
refresh(c, t)
|
||
# Autopilot: KYC без открытой вкладки + провижн после оплаты, даже если вкладку закрыли
|
||
aps = [r[0] for r in c.execute(
|
||
"select token from ap_orders where paid_notified=0 and created_at > datetime('now','-7 days') "
|
||
"and (pending_id!='' or kyc_status not in ('Approved','Declined','Abandoned','Expired'))")]
|
||
c.close()
|
||
for t in aps:
|
||
ap_status(t)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def clean(d, k, n=600):
|
||
return str(d.get(k) or "").strip()[:n]
|
||
|
||
|
||
def lab_apply(d, ip):
|
||
if clean(d, "company_hp"): # honeypot
|
||
return 200, {"ok": True, "token": "x" * 16}
|
||
if limited("apply", ip, 3):
|
||
return 429, {"error": "Too many applications from this network — try again in an hour."}
|
||
f = {k: clean(d, k) for k in ("name", "email", "role", "project_name", "project", "links", "stage",
|
||
"b_mission", "b_market", "b_product", "b_need",
|
||
"company_name", "company_country", "company_reg")}
|
||
f["email"] = f["email"].lower()
|
||
if not f["name"] or not f["project_name"] or not f["project"]:
|
||
return 400, {"error": "Name, lab name and a one-line description are required."}
|
||
if not EMAIL_RE.match(f["email"]):
|
||
return 400, {"error": "Please enter a valid email."}
|
||
if not d.get("consent"):
|
||
return 400, {"error": "Please confirm consent to identity verification."}
|
||
has_company = bool(f["company_name"])
|
||
links = f["links"]
|
||
if has_company:
|
||
links = (links + " | " if links else "") + \
|
||
f"company: {f['company_name']} ({f['company_country'] or '?'}, reg {f['company_reg'] or '?'})"
|
||
payload = {"name": f["name"], "role": f["role"] or "Founder", "project_name": f["project_name"],
|
||
"project": f["project"] + (f" [stage: {f['stage']}]" if f["stage"] else ""),
|
||
"links": links[:600], "contact": f["email"], "ref": "shkor",
|
||
"b_mission": f["b_mission"], "b_market": f["b_market"],
|
||
"b_product": f["b_product"], "b_need": f["b_need"]}
|
||
try:
|
||
req = urllib.request.Request(PORTAL_URL + "/api/studio/apply", data=json.dumps(payload).encode(),
|
||
headers={"Content-Type": "application/json", "Host": "42.cil.by",
|
||
"X-Forwarded-For": ip}, method="POST")
|
||
with urllib.request.urlopen(req, timeout=20) as r:
|
||
res = json.loads(r.read().decode())
|
||
except urllib.error.HTTPError as e:
|
||
try:
|
||
msg = json.loads(e.read().decode()).get("error")
|
||
except Exception:
|
||
msg = None
|
||
return (429 if e.code == 429 else 502), {"error": msg or "Couldn't register the application. Please try again."}
|
||
except Exception:
|
||
return 502, {"error": "Couldn't register the application. Please try again."}
|
||
if not res.get("ok"):
|
||
return 400, {"error": res.get("error") or "Application rejected."}
|
||
|
||
token = secrets.token_urlsafe(16)
|
||
c = db()
|
||
c.execute("insert into lab_apps(token, app_id, portal_track, name, email, project_name, company_name, "
|
||
"company_country, company_reg, ip) values (?,?,?,?,?,?,?,?,?,?)",
|
||
(token, res.get("id"), res.get("track_url") or "", f["name"], f["email"], f["project_name"],
|
||
f["company_name"], f["company_country"], f["company_reg"], ip))
|
||
callback = f"{PUBLIC_BASE}/apply/status/?t={token}"
|
||
kinds = [("kyc", KYC_WF)] + ([("kyb", KYB_WF)] if has_company else [])
|
||
for kind, wf in kinds:
|
||
sid = url = ""
|
||
status = "Not Started" if wf else "Not Enabled"
|
||
if wf:
|
||
try:
|
||
s = didit_session(wf, f"shkor-lab:{res.get('id')}:{kind}", callback)
|
||
sid, url, status = s.get("session_id", ""), s.get("url", ""), s.get("status", "Not Started")
|
||
except Exception:
|
||
status = "Error"
|
||
c.execute("insert into lab_checks(token, kind, session_id, url, status) values (?,?,?,?,?)",
|
||
(token, kind, sid, url, status))
|
||
c.commit()
|
||
kyc_url = c.execute("select url from lab_checks where token=? and kind='kyc'", (token,)).fetchone()["url"]
|
||
c.close()
|
||
return 200, {"ok": True, "token": token, "status_url": f"{PUBLIC_BASE}/apply/status/?t={token}", "kyc_url": kyc_url}
|
||
|
||
|
||
def portal(path, data=None, timeout=30):
|
||
"""Вызов портала (onboarding/Stripe/провижн) напрямую, с Host основного сайта."""
|
||
req = urllib.request.Request(PORTAL_URL + path, data=json.dumps(data).encode() if data is not None else None,
|
||
headers={"Content-Type": "application/json", "Host": "cil.by"},
|
||
method="POST" if data is not None else "GET")
|
||
with urllib.request.urlopen(req, timeout=timeout) as r:
|
||
return json.loads(r.read().decode())
|
||
|
||
|
||
def ap_refresh(c, o):
|
||
"""KYC-статус из Didit (пока не финальный); уведомление админу один раз."""
|
||
st = o["kyc_status"]
|
||
if o["kyc_session"] and st not in FINAL:
|
||
try:
|
||
st = didit_status(o["kyc_session"]) or st
|
||
except Exception:
|
||
return o
|
||
c.execute("update ap_orders set kyc_status=?, updated_at=datetime('now') where token=?", (st, o["token"]))
|
||
if st in FINAL and o["kyc_notified"] != st:
|
||
tg(f"{'✅' if st == 'Approved' else '⛔'} KYC {st} — Autopilot $99\n👤 {o['name']} · {o['email']}"
|
||
+ (f"\n🏢 {o['company']}" if o["company"] else ""))
|
||
c.execute("update ap_orders set kyc_notified=? where token=?", (st, o["token"]))
|
||
c.commit()
|
||
o = c.execute("select * from ap_orders where token=?", (o["token"],)).fetchone()
|
||
return o
|
||
|
||
|
||
def ap_start(d, ip):
|
||
if clean(d, "company_hp"): # honeypot
|
||
return 200, {"ok": True, "token": "x" * 16}
|
||
if limited("ap", ip, 5):
|
||
return 429, {"error": "Too many attempts from this network — try again in an hour."}
|
||
name, email, company = clean(d, "name", 120), clean(d, "email", 254).lower(), clean(d, "company", 160)
|
||
if not name:
|
||
return 400, {"error": "Please enter your full name (as in your ID)."}
|
||
if not EMAIL_RE.match(email):
|
||
return 400, {"error": "Please enter a valid email."}
|
||
if not d.get("consent"):
|
||
return 400, {"error": "Please accept the terms and identity verification."}
|
||
token = secrets.token_urlsafe(16)
|
||
try:
|
||
s = didit_session(KYC_WF, f"shkor-ap:{token}", f"{PUBLIC_BASE}/start/status/?t={token}")
|
||
except Exception:
|
||
return 502, {"error": "Identity verification is unavailable right now. Please try again in a few minutes."}
|
||
c = db()
|
||
c.execute("insert into ap_orders(token, name, email, company, ip, kyc_session, kyc_url, kyc_status) "
|
||
"values (?,?,?,?,?,?,?,?)", (token, name, email, company, ip, s.get("session_id", ""),
|
||
s.get("url", ""), s.get("status", "Not Started")))
|
||
c.commit()
|
||
c.close()
|
||
tg(f"🆕 Autopilot $99 — начат KYC\n👤 {name} · {email}" + (f"\n🏢 {company}" if company else ""))
|
||
return 200, {"ok": True, "token": token, "kyc_url": s.get("url", ""),
|
||
"status_url": f"{PUBLIC_BASE}/start/status/?t={token}"}
|
||
|
||
|
||
def ap_status(token):
|
||
c = db()
|
||
o = c.execute("select * from ap_orders where token=?", (token,)).fetchone()
|
||
if not o:
|
||
c.close()
|
||
return 404, {"error": "Order not found."}
|
||
o = ap_refresh(c, o)
|
||
out = {"ok": True, "name": o["name"], "email": o["email"], "kyc": o["kyc_status"]}
|
||
if o["kyc_status"] != "Approved":
|
||
out["stage"] = "declined" if o["kyc_status"] in FINAL else "kyc"
|
||
out["kyc_url"] = "" if o["kyc_status"] in FINAL else o["kyc_url"]
|
||
elif not o["pending_id"]:
|
||
out["stage"] = "pay"
|
||
else:
|
||
try:
|
||
with AP_LOCK:
|
||
r = portal(f"/api/onboarding/status?pending_id={o['pending_id']}", timeout=120)
|
||
except Exception:
|
||
r = {"status": "pending"}
|
||
if r.get("status") != "ready":
|
||
out["stage"] = "pay"
|
||
else:
|
||
if not o["paid_notified"]:
|
||
tg(f"💳 Autopilot $99 оплачен\n👤 {o['name']} · {o['email']}\n🧩 тенант {r.get('tenant_slug')}")
|
||
c.execute("update ap_orders set paid_notified=1 where token=?", (token,))
|
||
c.commit()
|
||
panel = r.get("panel")
|
||
out["stage"] = "ready" if panel else "deploying"
|
||
out["agent"] = r.get("agent")
|
||
out["portal_url"] = r.get("portal_url", "")
|
||
out["panel"] = panel
|
||
out["earnings"] = r.get("earnings")
|
||
c.close()
|
||
return 200, out
|
||
|
||
|
||
def ap_checkout(d):
|
||
token = clean(d, "t", 64)
|
||
c = db()
|
||
o = c.execute("select * from ap_orders where token=?", (token,)).fetchone()
|
||
if not o:
|
||
c.close()
|
||
return 404, {"error": "Order not found."}
|
||
o = ap_refresh(c, o)
|
||
if o["kyc_status"] != "Approved":
|
||
c.close()
|
||
return 403, {"error": "Payment opens after your identity is verified."}
|
||
try:
|
||
pid = o["pending_id"]
|
||
if not pid:
|
||
pid = portal("/api/onboarding/start", {"email": o["email"], "plan": "autopilot"})["pending_id"]
|
||
c.execute("update ap_orders set pending_id=? where token=?", (pid, token))
|
||
c.commit()
|
||
label = re.sub(r"[^a-z0-9]+", "-", (o["company"] or o["name"]).lower()).strip("-")[:16] or "autopilot"
|
||
portal("/api/onboarding/save", {"pending_id": pid, "company_name": o["company"] or o["name"],
|
||
"agent_name": f"{label}-{secrets.token_hex(2)}"})
|
||
r = portal("/api/onboarding/checkout", {"pending_id": pid, "plan": "autopilot",
|
||
"return_url": f"{PUBLIC_BASE}/start/status/?t={token}"})
|
||
except Exception:
|
||
c.close()
|
||
return 502, {"error": "Couldn't open the payment page. Please try again."}
|
||
c.close()
|
||
if not r.get("url"):
|
||
return 502, {"error": r.get("error") or "Couldn't open the payment page."}
|
||
return 200, {"ok": True, "url": r["url"]}
|
||
|
||
|
||
def lab_status(token):
|
||
c = db()
|
||
app = c.execute("select * from lab_apps where token=?", (token,)).fetchone()
|
||
if not app:
|
||
c.close()
|
||
return 404, {"error": "Application not found."}
|
||
refresh(c, token)
|
||
checks = {r["kind"]: {"status": r["status"], "url": r["url"] if r["status"] not in FINAL else ""}
|
||
for r in c.execute("select * from lab_checks where token=?", (token,))}
|
||
c.close()
|
||
return 200, {"ok": True, "app": {"id": app["app_id"], "name": app["name"], "project_name": app["project_name"],
|
||
"company_name": app["company_name"], "created_at": app["created_at"],
|
||
"track_url": app["portal_track"]},
|
||
"checks": checks}
|
||
|
||
|
||
class H(BaseHTTPRequestHandler):
|
||
def reply(self, code, obj):
|
||
body = json.dumps(obj).encode()
|
||
self.send_response(code)
|
||
self.send_header("Content-Type", "application/json")
|
||
self.send_header("Cache-Control", "no-store")
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
def ip(self):
|
||
return self.headers.get("X-Forwarded-For", self.client_address[0]).split(",")[0].strip()
|
||
|
||
def body(self):
|
||
n = int(self.headers.get("Content-Length") or 0)
|
||
if n > 16384:
|
||
return None
|
||
return self.rfile.read(n).decode("utf-8", "replace")
|
||
|
||
def do_GET(self):
|
||
u = urlsplit(self.path)
|
||
if u.path.rstrip("/") == "/api/lab/status":
|
||
t = (parse_qs(u.query).get("t") or [""])[0]
|
||
if not re.fullmatch(r"[A-Za-z0-9_-]{10,64}", t):
|
||
return self.reply(400, {"error": "Bad link."})
|
||
return self.reply(*lab_status(t))
|
||
if u.path.rstrip("/") == "/api/autopilot/status":
|
||
t = (parse_qs(u.query).get("t") or [""])[0]
|
||
if not re.fullmatch(r"[A-Za-z0-9_-]{10,64}", t):
|
||
return self.reply(400, {"error": "Bad link."})
|
||
return self.reply(*ap_status(t))
|
||
self.reply(404, {"error": "not found"})
|
||
|
||
def do_POST(self):
|
||
path = urlsplit(self.path).path.rstrip("/")
|
||
raw = self.body()
|
||
if raw is None:
|
||
return self.reply(413, {"error": "too large"})
|
||
if path == "/api/lab/apply":
|
||
try:
|
||
d = json.loads(raw or "{}")
|
||
except ValueError:
|
||
return self.reply(400, {"error": "bad json"})
|
||
return self.reply(*lab_apply(d, self.ip()))
|
||
if path in ("/api/autopilot/start", "/api/autopilot/checkout"):
|
||
try:
|
||
d = json.loads(raw or "{}")
|
||
except ValueError:
|
||
return self.reply(400, {"error": "bad json"})
|
||
return self.reply(*(ap_start(d, self.ip()) if path.endswith("start") else ap_checkout(d)))
|
||
if path != "/api/subscribe":
|
||
return self.reply(404, {"error": "not found"})
|
||
form = {k: v[0] for k, v in parse_qs(raw).items()}
|
||
if form.get("company"): # honeypot: pretend success to bots
|
||
return self.reply(200, {"ok": True})
|
||
if limited("sub", self.ip(), 10):
|
||
return self.reply(429, {"error": "Too many attempts — try again later."})
|
||
email = form.get("email", "").strip().lower()
|
||
if not EMAIL_RE.match(email) or len(email) > 254:
|
||
return self.reply(400, {"error": "Please enter a valid email."})
|
||
with db() as c:
|
||
c.execute("insert or ignore into subs(email, source, ip, ua) values (?,?,?,?)",
|
||
(email, form.get("source", "")[:32], self.ip(), self.headers.get("User-Agent", "")[:300]))
|
||
self.reply(200, {"ok": True})
|
||
|
||
def log_message(self, fmt, *args):
|
||
pass
|
||
|
||
|
||
if __name__ == "__main__":
|
||
os.makedirs(os.path.dirname(DB), exist_ok=True)
|
||
db().close()
|
||
threading.Thread(target=poller, daemon=True).start()
|
||
ThreadingHTTPServer(("127.0.0.1", PORT), H).serve_forever()
|