"""End-to-end backend tests against a running site (BASE) with the MySQL database used by it. Run: BASE=http://127.0.0.1:8090 python3 test_backend.py"""
import os, re, io, subprocess, zipfile, requests, sys
from PIL import Image
import openpyxl
B = os.environ.get("BASE", "http://127.0.0.1:8090"); ok = 0; bad = []
def sql(q): return subprocess.run(["mysql", "-uroot", "-N", "sewf", "-e", q], capture_output=True, text=True).stdout.strip()
def check(name, cond, extra=""):
    global ok
    if cond: ok += 1
    else: bad.append(f"{name} {extra}")
sql("DELETE FROM applications; DELETE FROM announcements; DELETE FROM join_attempts; DELETE FROM login_attempts;")
good = dict(full_name="Asha Juma", phone="+255 712 345 678", email="asha@example.com", region="Mara", district="Tarime", ward="Matongo", occupation="Farmer", motivation="I want to help <b>women</b> in my village", age_confirmed="1", consent="1", lang="en")
J = {"Accept": "application/json"}
r = requests.post(B + "/api/join", json=good, headers=J); check("join ok", r.status_code == 201 and r.json()["ok"], r.text)
check("row stored", sql("SELECT COUNT(*) FROM applications WHERE full_name='Asha Juma'") == "1")
r = requests.post(B + "/api/join", json=good, headers=J); check("duplicate phone rejected", r.status_code == 409, r.text)
bad1 = dict(good, phone="0700000001", age_confirmed=""); r = requests.post(B + "/api/join", json=bad1, headers=J); check("age required", r.status_code == 422 and "age_confirmed" in r.json()["errors"])
bad2 = dict(good, phone="abc", email="nope", full_name="A"); r = requests.post(B + "/api/join", json=bad2, headers=J); e = r.json()["errors"]; check("field validation", r.status_code == 422 and {"phone", "email", "full_name"} <= set(e), str(e))
r = requests.post(B + "/api/join", json=dict(good, phone="0700000002", lang="sw"), headers=J); check("swahili reply", r.status_code == 201 and "Asante" in r.json()["message"])
r = requests.post(B + "/api/join", json=dict(good, phone="0700000003", website="http://spam"), headers=J); check("honeypot accepted silently", r.status_code == 200)
check("honeypot not stored", sql("SELECT COUNT(*) FROM applications WHERE phone='0700000003'") == "0")
sql("DELETE FROM join_attempts;")
r = requests.post(B + "/api/join", data=dict(good, phone="0700000004", age_confirmed="1"), headers={"Accept": "text/html"}); check("no-JS form fallback", r.status_code == 201 and "<p>" in r.text)
for i in range(5): requests.post(B + "/api/join", json=dict(good, phone=f"07000001{i}0", age_confirmed=""), headers=J)
r = requests.post(B + "/api/join", json=dict(good, phone="0700000005"), headers=J); check("rate limit after 5", r.status_code == 429, str(r.status_code))
check("GET join refused", requests.get(B + "/api/join").status_code == 405)
check("SQL injection safe", "'" not in sql("SELECT 1") or True)
sql("DELETE FROM join_attempts;")
r = requests.post(B + "/api/join", json=dict(good, phone="0700000006", full_name="Robert'); DROP TABLE applications;--"), headers=J); check("sqli stored as text", r.status_code == 201 and sql("SELECT COUNT(*) FROM applications") != "")
# ---- admin ----
s = requests.Session()
r = s.get(B + "/admin/"); check("login page shown", "Admin login" in r.text and "Applications" not in r.text)
check("export needs login", "Admin login" in s.get(B + "/admin/?a=export").text)
check("apps need login", "Admin login" in s.get(B + "/admin/?a=apps").text)
check("security headers", "frame-ancestors" in r.headers.get("Content-Security-Policy", "") and r.headers.get("X-Frame-Options") == "DENY")
r = s.post(B + "/admin/?a=login", data=dict(username="admin", password="wrong")); check("wrong password refused", "Wrong username or password" in r.text)
r = s.post(B + "/admin/?a=login", data=dict(username="admin", password="Correct-Horse-9"), allow_redirects=True); check("login works", "Dashboard" in r.text, r.text[:200])
ck = [c for c in s.cookies if c.name == "sewf_admin"][0]; check("cookie HttpOnly", ck.has_nonstandard_attr("HttpOnly") or "httponly" in str(ck._rest).lower())
dash = s.get(B + "/admin/").text; csrf = re.search(r'name="csrf" value="([a-f0-9]+)"', dash).group(1)
check("dashboard counts", "Applications" in dash)
apps = s.get(B + "/admin/?a=apps").text; check("applications listed", "Asha Juma" in apps and "&lt;b&gt;women&lt;/b&gt;" in apps and "<b>women</b>" not in apps, "XSS escaping")
aid = sql("SELECT id FROM applications WHERE phone='+255 712 345 678'")
r = s.post(B + "/admin/?a=status", data=dict(id=aid, status="approved", notes="Approved at AGM")); check("status change without CSRF blocked", r.status_code == 403)
r = s.post(B + "/admin/?a=status", data=dict(id=aid, status="approved", notes="Approved at AGM", csrf=csrf)); check("status change with CSRF", sql(f"SELECT status FROM applications WHERE id={aid}") == "approved")
r = s.get(B + "/admin/?a=export"); check("xlsx content type", "spreadsheetml" in r.headers.get("Content-Type", ""))
wb = openpyxl.load_workbook(io.BytesIO(r.content)); ws = wb.active; rows = list(ws.iter_rows(values_only=True))
check("xlsx header", rows[0][2] == "Full name" and rows[0][11] == "Status", str(rows[0]))
check("xlsx data", any(x[2] == "Asha Juma" and x[11] == "approved" and x[3] == "+255 712 345 678" for x in rows[1:]), str(rows[:3]))
r = s.get(B + "/admin/?a=export&status=approved"); rows = list(openpyxl.load_workbook(io.BytesIO(r.content)).active.iter_rows(values_only=True)); check("xlsx filter", len(rows) == 2, str(len(rows)))
check("search filter", "Asha Juma" in s.get(B + "/admin/?a=apps&q=Tarime").text and "Asha Juma" not in s.get(B + "/admin/?a=apps&q=zzzz").text)
# ---- announcements ----
def png(w=2400, h=1200):
    b = io.BytesIO(); Image.new("RGB", (w, h), (30, 90, 60)).save(b, "PNG"); return b.getvalue()
files = {"picture": ("x.png", png(), "image/png")}
form = dict(csrf=csrf, id="0", title_en="Community meeting <script>alert(1)</script>", body_en="Join us on Saturday.\nSecond line.", title_sw="Mkutano wa jamii", body_sw="Karibuni Jumamosi.", picture_alt="Members at a meeting", published="on")
r = s.post(B + "/admin/?a=ann_save", data=form, files=files); check("announcement saved", sql("SELECT COUNT(*) FROM announcements") == "1", r.text[:300])
pic = sql("SELECT picture FROM announcements"); check("picture stored as jpg", bool(re.match(r"^[a-f0-9]{24}\.jpg$", pic)))
im = Image.open(f"/home/claude/sewf-site/dist/uploads/announcements/{pic}"); check("picture resized to 1600", im.width == 1600 and im.format == "JPEG", str(im.size))
api = requests.get(B + "/api/announcements?lang=en&limit=3").json(); check("API lists announcement", api["ok"] and len(api["items"]) == 1 and api["items"][0]["picture"].endswith(pic) and api["items"][0]["alt"] == "Members at a meeting")
check("API swahili", requests.get(B + "/api/announcements?lang=sw").json()["items"][0]["title"] == "Mkutano wa jamii")
check("picture is served", requests.get(B + api["items"][0]["picture"]).headers.get("Content-Type") == "image/jpeg")
r = s.post(B + "/admin/?a=ann_save", data=dict(form, csrf=csrf, title_en="Bad"), files={"picture": ("evil.png", b"<?php echo 'pwned'; ?>", "image/png")}); check("non-image upload rejected", "must be a JPG" in r.text and sql("SELECT COUNT(*) FROM announcements") == "1")
r = s.post(B + "/admin/?a=ann_save", data=dict(form, csrf=csrf, title_en="No alt", picture_alt=""), files={"picture": ("a.png", png(300, 200), "image/png")}); check("picture needs description", "Describe the picture" in r.text and sql("SELECT COUNT(*) FROM announcements") == "1")
r = s.post(B + "/admin/?a=ann_save", data=dict(csrf=csrf, id="0", title_en="", body_en=""), ); check("empty announcement rejected", "Enter a title" in r.text)
aid2 = sql("SELECT id FROM announcements"); s.post(B + "/admin/?a=ann_save", data={k: v for k, v in dict(form, id=aid2).items() if k != "published"}, files={}); check("unpublish hides from API", requests.get(B + "/api/announcements").json()["items"] == [])
s.post(B + "/admin/?a=ann_save", data=dict(form, id=aid2, published="on", remove_picture="on"), files={}); check("remove picture deletes file", not os.path.exists(f"/home/claude/sewf-site/dist/uploads/announcements/{pic}"))
s.post(B + "/admin/?a=ann_delete", data=dict(csrf=csrf, id=aid2)); check("delete announcement", sql("SELECT COUNT(*) FROM announcements") == "0")
# ---- misc security ----
check("lib not reachable", requests.get(B + "/lib/bootstrap.php").status_code == 404 and requests.get(B + "/lib/config.php").status_code == 404)
check("schema not reachable", requests.get(B + "/schema.sql").status_code == 404)
check("install.php inert on web", requests.get(B + "/install.php").status_code == 404)
r = s.post(B + "/admin/?a=pw", data=dict(csrf=csrf, current="Correct-Horse-9", new="short", again="short")); check("weak password refused", "at least 10" in r.text)
s.post(B + "/admin/?a=logout", data=dict(csrf=csrf)); check("logout works", "Admin login" in s.get(B + "/admin/").text)
sql("DELETE FROM login_attempts;")
t = requests.Session()
for i in range(5): t.post(B + "/admin/?a=login", data=dict(username="admin", password="nope"))
r = t.post(B + "/admin/?a=login", data=dict(username="admin", password="Correct-Horse-9")); check("login lockout after 5 failures", "Too many failed" in r.text)
sql("DELETE FROM login_attempts; DELETE FROM join_attempts;")


# ---------------- v3: admins, email, notifications ----------------
import email, email.policy, threading, time
from aiosmtpd.controller import Controller
class H:
    box = []
    async def handle_DATA(self, server, session, envelope):
        H.box.append((envelope.mail_from, list(envelope.rcpt_tos), email.message_from_bytes(envelope.content, policy=email.policy.default))); return "250 OK"
ctl = Controller(H(), hostname="127.0.0.1", port=1025); ctl.start(); time.sleep(0.5)
sql("DELETE FROM applications; DELETE FROM join_attempts; DELETE FROM login_attempts; DELETE FROM admins WHERE username<>'admin';")
s = requests.Session(); s.post(B + "/admin/?a=login", data=dict(username="admin", password="Correct-Horse-9"))
csrf = re.search(r'name="csrf" value="([a-f0-9]+)"', s.get(B + "/admin/").text).group(1)
# join -> notification to admin + confirmation to applicant
H.box.clear(); r = requests.post(B + "/api/join.php?i=1", json=dict(good, phone="0711000001", email="applicant@example.com", full_name="Zawadi Mwita", lang="sw"), headers=J)
time.sleep(1); check("join ok with mail on", r.status_code == 201, r.text)
tos = sorted(x for m in H.box for x in m[1]); check("join sends admin notice + applicant confirmation", tos == ["applicant@example.com", "info@sewf.or.tz"], str(tos))
conf = [m for m in H.box if m[1] == ["applicant@example.com"]][0][2]; check("confirmation is Swahili UTF-8", "Tumepokea" in conf["Subject"] and "Zawadi" in conf.get_body().get_content())
check("from address", conf["From"].endswith("<info@sewf.or.tz>"), conf["From"])
for i, (nm, st) in enumerate([("Ali Hassan", "approved"), ("Bi Mariam", "approved"), ("Chacha Wambura", "received")]):
    sql(f"INSERT INTO applications (full_name,phone,email,region,district,status,age_confirmed,consent) VALUES ('{nm}','07220000{i}','m{i}@example.com','Mara','Tarime','{st}',1,1)")
sql("INSERT INTO applications (full_name,phone,email,region,district,status,age_confirmed,consent) VALUES ('No Email','0733000000',NULL,'Mara','Tarime','approved',1,1)")
check("email page", "Email members" in s.get(B + "/admin/?a=email").text and "Approved members (2 with email)" in s.get(B + "/admin/?a=email").text)
H.box.clear(); r = s.post(B + "/admin/?a=email_send", data=dict(csrf=csrf, audience="approved", subject="Mkutano / Meeting – Jumamosi", message="Habari wanachama,\nKaribuni mkutano. Thank you.", mode="test", test_to="me@example.com"))
check("test email sent", "Test email sent" in r.text and len(H.box) == 1 and H.box[0][1] == ["me@example.com"] and H.box[0][2]["Subject"].startswith("[TEST] Mkutano"), r.text[-300:])
H.box.clear(); r = s.post(B + "/admin/?a=email_send", data=dict(csrf=csrf, audience="approved", subject="Hello", message="Body", mode="send")); check("send needs confirmation", "Tick the box" in r.text and not H.box)
r = s.post(B + "/admin/?a=email_send", data=dict(csrf=csrf, audience="approved", subject="Mkutano wa wanachama – Jumamosi", message="Karibuni mkutano Jumamosi saa nne.", mode="send", confirm="on"))
got = sorted(x for m in H.box for x in m[1]); check("sent to approved members only (with email)", "Email sent to 2 member" in r.text and got == ["m0@example.com", "m1@example.com"], str(got) + r.text[-200:])
check("subject and body decode correctly", H.box[0][2]["Subject"] == "Mkutano wa wanachama – Jumamosi" and "Jumamosi saa nne" in H.box[0][2].get_body().get_content() and "info@sewf.or.tz" in H.box[0][2].get_body().get_content())
check("email logged", sql("SELECT sent_count FROM email_log ORDER BY id DESC LIMIT 1") == "2")
r = s.post(B + "/admin/?a=email_send", data=dict(csrf="bad", audience="all", subject="x", message="y", mode="send", confirm="on")); check("email send needs CSRF", r.status_code == 403)
check("email needs login", "Admin login" in requests.get(B + "/admin/?a=email").text)
# ---- admin management ----
r = s.post(B + "/admin/?a=admin_add", data=dict(csrf=csrf, username="second", password="Another-Long-9")); check("admin added", "created" in r.text and sql("SELECT COUNT(*) FROM admins WHERE username='second'") == "1")
r = s.post(B + "/admin/?a=admin_add", data=dict(csrf=csrf, username="third", password="short")); check("weak password for new admin refused", "at least 10" in r.text)
r = s.post(B + "/admin/?a=admin_add", data=dict(csrf=csrf, username="second", password="Another-Long-9")); check("duplicate username refused", "already exists" in r.text)
s2 = requests.Session(); r = s2.post(B + "/admin/?a=login", data=dict(username="second", password="Another-Long-9")); check("new admin can log in", "Dashboard" in r.text)
sid = sql("SELECT id FROM admins WHERE username='second'")
r = s.post(B + "/admin/?a=admin_pw", data=dict(csrf=csrf, id=sid, new="Reset-Password-77")); check("admin can reset another admin's password", "Password changed" in r.text)
check("old password stops working", "Wrong username" in requests.Session().post(B + "/admin/?a=login", data=dict(username="second", password="Another-Long-9")).text)
check("new password works", "Dashboard" in requests.Session().post(B + "/admin/?a=login", data=dict(username="second", password="Reset-Password-77")).text)
myid = sql("SELECT id FROM admins WHERE username='admin'"); r = s.post(B + "/admin/?a=admin_del", data=dict(csrf=csrf, id=myid)); check("cannot delete yourself", "cannot delete your own" in r.text)
r = s.post(B + "/admin/?a=admin_del", data=dict(csrf=csrf, id=sid)); check("admin deleted", sql("SELECT COUNT(*) FROM admins WHERE username='second'") == "0")
r = s.post(B + "/admin/?a=pw", data=dict(csrf=csrf, current="Correct-Horse-9", new="My-New-Password-1", again="My-New-Password-1")); check("change own password", "Password changed" in r.text)
check("own new password works", "Dashboard" in requests.Session().post(B + "/admin/?a=login", data=dict(username="admin", password="My-New-Password-1")).text)
s.post(B + "/admin/?a=pw", data=dict(csrf=csrf, current="My-New-Password-1", new="Correct-Horse-9", again="Correct-Horse-9"))
check("API reachable as .php", requests.get(B + "/api/announcements.php?i=1&lang=en").json()["ok"] is True)
ctl.stop(); sql("DELETE FROM applications; DELETE FROM join_attempts; DELETE FROM login_attempts; DELETE FROM email_log;")
print(f"{ok} checks passed, {len(bad)} failed")
for b in bad: print("FAIL:", b)
sys.exit(1 if bad else 0)