161 lines
7.8 KiB
Python
Executable File
161 lines
7.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""End-to-end API check of a running PostfixAdmin-RS instance.
|
|
|
|
python3 scripts/api-smoke.py http://127.0.0.1:8080 admin@example.test admin123
|
|
|
|
Uses a global administrator, creates the domains rs-smoke-a.test and
|
|
rs-smoke-b.test, exercises the PostfixAdmin business rules (limits, quota,
|
|
aliases, forwarding, vacation, permissions) and deletes everything again.
|
|
Run it only against a development or staging database. With
|
|
features.emailcheck_resolve_domain = true the .test domains are rejected,
|
|
disable that check for the test instance.
|
|
"""
|
|
import http.cookiejar
|
|
import json
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
BASE = sys.argv[1].rstrip("/") if len(sys.argv) > 1 else "http://127.0.0.1:8080"
|
|
ADMIN = sys.argv[2] if len(sys.argv) > 2 else "admin@example.test"
|
|
PASSWORD = sys.argv[3] if len(sys.argv) > 3 else "admin123"
|
|
A, B = "rs-smoke-a.test", "rs-smoke-b.test"
|
|
failures = 0
|
|
|
|
|
|
class Client:
|
|
def __init__(self):
|
|
self.jar = http.cookiejar.CookieJar()
|
|
self.opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self.jar))
|
|
self.csrf = ""
|
|
|
|
def req(self, method, path, body=None, expect=None, quiet=False, silent=False):
|
|
global failures
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
request = urllib.request.Request(BASE + "/api" + path, data=data, method=method)
|
|
if data:
|
|
request.add_header("content-type", "application/json")
|
|
if self.csrf:
|
|
request.add_header("x-csrf-token", self.csrf)
|
|
try:
|
|
response = self.opener.open(request)
|
|
code, text = response.status, response.read().decode()
|
|
except urllib.error.HTTPError as error:
|
|
code, text = error.code, error.read().decode()
|
|
ok = (expect is None and code < 300) or code == expect
|
|
if silent:
|
|
return json.loads(text) if text else None
|
|
if not ok:
|
|
failures += 1
|
|
if not ok or not quiet:
|
|
detail = "" if ok and code < 300 else " " + text[:160]
|
|
print(f"{'ok ' if ok else 'FAIL'} {method:6} {path} -> {code}{detail}")
|
|
return json.loads(text) if text else None
|
|
|
|
def login(self, username, password, kind=None):
|
|
body = {"username": username, "password": password}
|
|
if kind:
|
|
body["kind"] = kind
|
|
result = self.req("POST", "/auth/login", body)
|
|
self.csrf = (result or {}).get("csrf", "")
|
|
return result
|
|
|
|
|
|
def main():
|
|
global failures
|
|
admin = Client()
|
|
admin.login(ADMIN, PASSWORD)
|
|
runtime = admin.req("GET", "/system/features", quiet=True)
|
|
features = runtime["features"]
|
|
quota = features.get("quota") and features.get("domain_quota")
|
|
|
|
# remove leftovers of an interrupted run (result ignored)
|
|
admin.req("DELETE", f"/admins/da@{A}", silent=True)
|
|
admin.req("DELETE", f"/alias-domains/{B}", silent=True)
|
|
for domain in (A, B):
|
|
admin.req("DELETE", f"/domains/{domain}", silent=True)
|
|
admin.req("POST", "/domains", {"domain": A, "description": "smoke", "aliases": 6, "mailboxes": 2,
|
|
"maxquota": 100, "quota": 150, "default_aliases": True})
|
|
admin.req("POST", "/domains", {"domain": B, "aliases": 0, "mailboxes": 0, "maxquota": 0, "quota": 0,
|
|
"default_aliases": False})
|
|
admin.req("POST", "/domains", {"domain": A}, expect=409)
|
|
admin.req("POST", "/domains", {"domain": "bad-limit.test", "aliases": -2}, expect=400)
|
|
|
|
pw = "Smoke12pass"
|
|
if quota:
|
|
admin.req("POST", "/mailboxes", {"domain": A, "local_part": "u1", "password": pw, "quota": 0}, expect=400)
|
|
admin.req("POST", "/mailboxes", {"domain": A, "local_part": "u1", "password": pw, "quota": 101}, expect=400)
|
|
admin.req("POST", "/mailboxes", {"domain": A, "local_part": "u1", "password": pw, "quota": 100, "name": "User One"})
|
|
if quota:
|
|
admin.req("POST", "/mailboxes", {"domain": A, "local_part": "u2", "password": pw, "quota": 60}, expect=400)
|
|
hint = admin.req("GET", f"/mailboxes/quota-hint?domain={A}", quiet=True)
|
|
if hint["allowed_mb"] != 50:
|
|
print("FAIL quota hint", hint)
|
|
admin.req("POST", "/mailboxes", {"domain": A, "local_part": "u2", "password": pw, "quota": 50})
|
|
admin.req("POST", "/mailboxes", {"domain": A, "local_part": "u3", "password": pw, "quota": 0}, expect=409)
|
|
admin.req("POST", "/mailboxes", {"domain": B, "local_part": "weak", "password": "abc"}, expect=400)
|
|
|
|
# aliases (4 default aliases + limit 6)
|
|
admin.req("POST", "/aliases", {"domain": A, "local_part": "info", "goto": [f"u1@{A}"]})
|
|
admin.req("POST", "/aliases", {"domain": A, "local_part": "sales", "goto": [f"u1@{A}"]})
|
|
admin.req("POST", "/aliases", {"domain": A, "local_part": "more", "goto": [f"u1@{A}"]}, expect=409)
|
|
admin.req("POST", "/aliases", {"domain": B, "local_part": "loop", "goto": [f"loop@{B}"]}, expect=400)
|
|
admin.req("POST", "/aliases", {"domain": B, "local_part": "*", "goto": [f"@{A}"]})
|
|
admin.req("POST", "/aliases", {"domain": B, "local_part": "root", "goto": ["root"]})
|
|
listing = admin.req("GET", f"/aliases?domain={A}", quiet=True)
|
|
if any(item["is_mailbox"] for item in listing["items"]):
|
|
print("FAIL mailbox aliases must not be listed for admins")
|
|
failures += 1
|
|
|
|
# forwarding / activity / search
|
|
admin.req("PUT", f"/mailboxes/u1@{A}/forwarding", {"forward_to": ["ext@example.org"], "deliver_to_mailbox": True})
|
|
mailbox = admin.req("GET", f"/mailboxes/u1@{A}", quiet=True)
|
|
if mailbox["forward_to"] != ["ext@example.org"] or not mailbox["deliver_to_mailbox"]:
|
|
print("FAIL forwarding", mailbox)
|
|
admin.req("POST", f"/aliases/sales@{A}/active", {"active": False})
|
|
admin.req("POST", f"/aliases/delete-inactive/{A}")
|
|
found = admin.req("GET", f"/mailboxes?domain={A}&q=one", quiet=True)
|
|
if found["total"] != 1:
|
|
print("FAIL search", found)
|
|
|
|
if features.get("vacation"):
|
|
admin.req("PUT", f"/vacation/u1@{A}", {"email": f"u1@{A}", "subject": "Away", "body": "Back soon",
|
|
"interval_time": (runtime["defaults"].get("vacation_choice_of_reply") or [0])[0],
|
|
"active_from": "2026-01-01", "active_until": "2026-01-31", "active": True})
|
|
mailbox = admin.req("GET", f"/mailboxes/u1@{A}", quiet=True)
|
|
if not mailbox["vacation_active"]:
|
|
print("FAIL vacation flag")
|
|
admin.req("DELETE", f"/vacation/u1@{A}")
|
|
|
|
# domain administrator
|
|
admin.req("POST", "/admins", {"username": f"da@{A}", "password": pw, "domains": [A]})
|
|
dadmin = Client()
|
|
dadmin.login(f"da@{A}", pw)
|
|
dadmin.req("POST", "/domains", {"domain": "x.test"}, expect=403)
|
|
dadmin.req("GET", f"/mailboxes?domain={B}", expect=403)
|
|
dadmin.req("POST", f"/mailboxes/u2@{A}/active", {"active": False})
|
|
admin.req("PUT", f"/admins/da@{A}", {"username": f"da@{A}", "password": "Changed12pw", "domains": [A], "active": True})
|
|
dadmin.req("GET", "/domains", expect=401)
|
|
|
|
# mailbox user
|
|
user = Client()
|
|
user.login(f"u1@{A}", pw)
|
|
user.req("GET", f"/mailboxes/u2@{A}", expect=403)
|
|
user.req("PUT", f"/mailboxes/u1@{A}/forwarding", {"forward_to": [], "deliver_to_mailbox": True})
|
|
user.req("POST", "/security/password", {"current_password": pw, "new_password": pw}, expect=400)
|
|
user.req("POST", "/security/password", {"current_password": pw, "new_password": "Newer12pass"})
|
|
|
|
# alias domain + deletion rules
|
|
admin.req("POST", "/alias-domains", {"alias_domain": B, "target_domain": A})
|
|
admin.req("DELETE", f"/domains/{A}", expect=409)
|
|
admin.req("DELETE", f"/alias-domains/{B}")
|
|
admin.req("DELETE", f"/admins/da@{A}")
|
|
admin.req("DELETE", f"/domains/{B}")
|
|
admin.req("DELETE", f"/domains/{A}")
|
|
print(f"\n{failures} failure(s)")
|
|
sys.exit(1 if failures else 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|