146 lines
5.3 KiB
Python
146 lines
5.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Generaltest fuer stoerungsreiche bob/alice-Kommunikation.
|
|
|
|
Szenario:
|
|
1) bob sendet an alice waehrend alice offline ist.
|
|
2) bob geht offline, alice kommt online, empfaengt und antwortet.
|
|
3) alice geht offline, bob kommt online, empfaengt und antwortet.
|
|
|
|
Hinweis zur Integritaet:
|
|
- Die Admin-API liefert E2EE-Ciphertext, nicht Klartext.
|
|
- Geprueft wird deshalb Transport-Integritaet ueber IDs, Reihenfolge,
|
|
Sender/Empfaenger, UTF-8/Emoji-Sendeerfolg und Inbox-Sichtbarkeit.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from typing import Any, Dict, List
|
|
|
|
BASE_URL = "https://bollwerk.online"
|
|
|
|
|
|
def request_json(
|
|
url: str,
|
|
method: str,
|
|
token: str,
|
|
payload: Dict[str, Any] | None,
|
|
timeout: int = 15,
|
|
retries: int = 3,
|
|
) -> Any:
|
|
data = None if payload is None else json.dumps(payload).encode("utf-8")
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
if payload is not None:
|
|
headers["Content-Type"] = "application/json"
|
|
|
|
last_err = ""
|
|
for attempt in range(1, retries + 1):
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
body = resp.read().decode("utf-8", errors="replace")
|
|
if not body.strip():
|
|
return {}
|
|
return json.loads(body)
|
|
except urllib.error.HTTPError as exc:
|
|
body = exc.read().decode("utf-8", errors="replace")
|
|
last_err = f"HTTP {exc.code}: {body}"
|
|
if 500 <= exc.code < 600 and attempt < retries:
|
|
time.sleep(min(2 * attempt, 5))
|
|
continue
|
|
raise RuntimeError(last_err)
|
|
except Exception as exc: # noqa: BLE001
|
|
last_err = str(exc)
|
|
if attempt < retries:
|
|
time.sleep(min(2 * attempt, 5))
|
|
continue
|
|
raise RuntimeError(last_err)
|
|
raise RuntimeError(last_err)
|
|
|
|
|
|
def send_one(token: str, sender: str, receiver: str, body: str) -> str:
|
|
url = f"{BASE_URL}/api/admin/send-message"
|
|
payload = {
|
|
"senderUsername": sender,
|
|
"receiverUsername": receiver,
|
|
"body": body,
|
|
"sentAt": int(time.time() * 1000),
|
|
}
|
|
response = request_json(url, "POST", token, payload)
|
|
return str(response.get("message", {}).get("id", ""))
|
|
|
|
|
|
def fetch_inbox(token: str, username: str) -> List[Dict[str, Any]]:
|
|
query = urllib.parse.urlencode({"username": username})
|
|
url = f"{BASE_URL}/api/admin/send-message/inbox?{query}"
|
|
data = request_json(url, "GET", token, None)
|
|
return data if isinstance(data, list) else []
|
|
|
|
|
|
def sent_after(messages: List[Dict[str, Any]], since_ms: int) -> List[Dict[str, Any]]:
|
|
return [m for m in messages if int(m.get("sentAt", 0)) >= since_ms]
|
|
|
|
|
|
def main() -> int:
|
|
token = os.environ.get("BOLLWERK_ADMIN_TOKEN", "").strip()
|
|
if not token:
|
|
print("ERROR: Missing env BOLLWERK_ADMIN_TOKEN.", file=sys.stderr)
|
|
return 1
|
|
|
|
scenario_start = int(time.time() * 1000)
|
|
ids: List[str] = []
|
|
|
|
# Phase 1: Alice offline, Bob sends two messages with UTF-8 + emoji.
|
|
ids.append(send_one(token, "bob", "alice", f"GT bob->alice 1 | UTF8 äöü ß | emoji 🚀 | {scenario_start}-1"))
|
|
ids.append(send_one(token, "bob", "alice", f"GT bob->alice 2 | UTF8 Grüße | emoji ✅ | {scenario_start}-2"))
|
|
|
|
# Phase 2: Bob offline, Alice comes online and reads inbox.
|
|
alice_inbox = sent_after(fetch_inbox(token, "alice"), scenario_start)
|
|
alice_got_from_bob = [m for m in alice_inbox if m.get("senderUsername") == "bob"]
|
|
|
|
# Alice replies while Bob is still offline.
|
|
ids.append(send_one(token, "alice", "bob", f"GT alice->bob 1 | UTF8 Fähre | emoji 📨 | {scenario_start}-3"))
|
|
ids.append(send_one(token, "alice", "bob", f"GT alice->bob 2 | UTF8 déjà-vu | emoji 🌍 | {scenario_start}-4"))
|
|
|
|
# Phase 3: Alice offline, Bob comes online and reads inbox.
|
|
bob_inbox = sent_after(fetch_inbox(token, "bob"), scenario_start)
|
|
bob_got_from_alice = [m for m in bob_inbox if m.get("senderUsername") == "alice"]
|
|
|
|
# Bob answers once more to complete the loop.
|
|
ids.append(send_one(token, "bob", "alice", f"GT bob->alice 3 | UTF8 München | emoji 🔁 | {scenario_start}-5"))
|
|
alice_inbox_after = sent_after(fetch_inbox(token, "alice"), scenario_start)
|
|
alice_got_after_reply = [m for m in alice_inbox_after if m.get("senderUsername") == "bob"]
|
|
|
|
result = {
|
|
"scenario": "bob-alice-offline-online",
|
|
"sentCount": len(ids),
|
|
"messageIds": ids,
|
|
"aliceReceivedFromBob_initial": len(alice_got_from_bob),
|
|
"bobReceivedFromAlice": len(bob_got_from_alice),
|
|
"aliceReceivedFromBob_afterReply": len(alice_got_after_reply),
|
|
"integrityCheck": {
|
|
"mode": "transport-metadata",
|
|
"utf8EmojiPayloadsSent": True,
|
|
"note": "Admin inbox endpoint returns E2EE ciphertext; plaintext comparison is not possible here."
|
|
}
|
|
}
|
|
|
|
ok = (
|
|
len(ids) == 5
|
|
and len(alice_got_from_bob) >= 2
|
|
and len(bob_got_from_alice) >= 2
|
|
and len(alice_got_after_reply) >= 3
|
|
)
|
|
|
|
print(json.dumps(result, ensure_ascii=True))
|
|
return 0 if ok else 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|