292 lines
13 KiB
Python
292 lines
13 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)
|
|
|
|
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]
|
|
|
|
|
|
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 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(600)
|
|
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)
|
|
c.close()
|
|
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 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))
|
|
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 != "/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()
|