64 lines
2.6 KiB
Python
64 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""shkor.com email capture: POST /api/subscribe -> SQLite. Stdlib only.
|
|
|
|
Env: SUBS_DB (default /var/lib/shkor-subscribe/subs.db), PORT (default 8097).
|
|
Export: sqlite3 $SUBS_DB "select email,source,created_at from subs"
|
|
"""
|
|
import json, os, re, sqlite3, time
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import parse_qs
|
|
|
|
DB = os.environ.get("SUBS_DB", "/var/lib/shkor-subscribe/subs.db")
|
|
PORT = int(os.environ.get("PORT", "8097"))
|
|
EMAIL_RE = re.compile(r"^[^@\s]{1,64}@[^@\s]+\.[A-Za-z]{2,}$")
|
|
RATE = {} # ip -> [timestamps]
|
|
|
|
|
|
def db():
|
|
c = sqlite3.connect(DB)
|
|
c.execute("""create table if not exists subs (
|
|
email text primary key, source text, ip text, ua text,
|
|
created_at text default (datetime('now')))""")
|
|
return c
|
|
|
|
|
|
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("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_POST(self):
|
|
if self.path.rstrip("/") != "/api/subscribe":
|
|
return self.reply(404, {"error": "not found"})
|
|
n = int(self.headers.get("Content-Length") or 0)
|
|
if n > 4096:
|
|
return self.reply(413, {"error": "too large"})
|
|
form = {k: v[0] for k, v in parse_qs(self.rfile.read(n).decode("utf-8", "replace")).items()}
|
|
if form.get("company"): # honeypot: pretend success to bots
|
|
return self.reply(200, {"ok": True})
|
|
ip = self.headers.get("X-Forwarded-For", self.client_address[0]).split(",")[0].strip()
|
|
now = time.time()
|
|
hits = [t for t in RATE.get(ip, []) if now - t < 3600]
|
|
if len(hits) >= 10:
|
|
return self.reply(429, {"error": "Too many attempts — try again later."})
|
|
RATE[ip] = hits + [now]
|
|
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], 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()
|
|
ThreadingHTTPServer(("127.0.0.1", PORT), H).serve_forever()
|