davidandClaude Opus 5.5
ce7cf08d60
server/report-receiver/receiver.py takes what report sends to federated.sh/report, behind Caddy, and files each report as an issue in the private Federated_Computer/coredesktop-reports repository with the full log attached. It saves each report before replying and tries Gitea again later if it's down, limits how many reports one address can send, and refuses oversized logs, gzip bombs included. Standard library only. The README has the deployment steps: a Gitea user that can only file issues there, the systemd unit, and the Caddy route. make test runs its tests, which send reports with the real report client. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
213 lines
9.5 KiB
Python
213 lines
9.5 KiB
Python
"""Tests for receiver.py. Run with: python3 -I -m unittest discover -s server/report-receiver
|
|
|
|
They start the receiver on a free port with a fake Gitea, and send it reports
|
|
with the real `report` client, the way a CoreDesktop machine would."""
|
|
|
|
import gzip
|
|
import http.server
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
import receiver # noqa: E402
|
|
|
|
REPO = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
class FakeGitea:
|
|
"""Records the issues and attachments it's given; fail=True answers 500."""
|
|
|
|
def __init__(self):
|
|
self.issues = []
|
|
self.attachments = []
|
|
self.fail = False
|
|
fake = self
|
|
|
|
class Handler(http.server.BaseHTTPRequestHandler):
|
|
def do_POST(self):
|
|
data = self.rfile.read(int(self.headers["Content-Length"]))
|
|
if fake.fail:
|
|
self.send_response(500)
|
|
self.end_headers()
|
|
return
|
|
if self.path.endswith("/issues"):
|
|
fake.issues.append({"path": self.path, "auth": self.headers["Authorization"], **json.loads(data)})
|
|
reply = {"number": len(fake.issues)}
|
|
else:
|
|
fields = receiver.parse_multipart(self.headers["Content-Type"], data)
|
|
fake.attachments.append({"path": self.path, "text": fields["attachment"].decode()})
|
|
reply = {"id": len(fake.attachments)}
|
|
body = json.dumps(reply).encode()
|
|
self.send_response(201)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
|
|
self.url = f"http://127.0.0.1:{self.server.server_port}"
|
|
threading.Thread(target=self.server.serve_forever, daemon=True).start()
|
|
|
|
def close(self):
|
|
self.server.shutdown()
|
|
|
|
|
|
class ReceiverTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = Path(tempfile.mkdtemp())
|
|
self.gitea = FakeGitea()
|
|
self.store = receiver.Store(self.tmp / "state", receiver.Gitea(self.gitea.url, "Federated_Computer/coredesktop-reports", "secret"))
|
|
self.limiter = receiver.RateLimiter()
|
|
handler = receiver.make_handler(self.store, self.limiter)
|
|
handler.log_message = lambda *args: None
|
|
self.server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler)
|
|
self.url = f"http://127.0.0.1:{self.server.server_port}/report"
|
|
threading.Thread(target=self.server.serve_forever, daemon=True).start()
|
|
|
|
# A CoreDesktop home with one failed update's log
|
|
self.home = self.tmp / "home"
|
|
logs = self.home / ".local/state/coredesktop/logs"
|
|
logs.mkdir(parents=True)
|
|
self.log = logs / "update-2026-11-08-110307.log"
|
|
self.log.write_text(f"Updating CoreDesktop...\nCloning into '{self.home}/x'...\n"
|
|
"Done, with errors in: 71-siblings.sh\n")
|
|
|
|
def tearDown(self):
|
|
self.server.shutdown()
|
|
self.gitea.close()
|
|
shutil.rmtree(self.tmp)
|
|
|
|
def send(self, *args):
|
|
"""Run the report client, answering yes, and return its output."""
|
|
tty = self.tmp / "tty"
|
|
tty.write_text("y\n")
|
|
env = {"HOME": str(self.home), "USER": "alice", "PATH": os.environ["PATH"],
|
|
"TMPDIR": str(self.tmp), "COREDESKTOP_REPORT_URL": self.url, "COREDESKTOP_TTY": str(tty)}
|
|
result = subprocess.run(["bash", str(REPO / "report"), *args], env=env,
|
|
capture_output=True, text=True, timeout=60)
|
|
return result.stdout + result.stderr
|
|
|
|
def wait_for(self, condition):
|
|
for _ in range(100):
|
|
if condition():
|
|
return
|
|
time.sleep(0.05)
|
|
self.fail("timed out")
|
|
|
|
def test_a_report_from_the_client_is_filed_with_its_log(self):
|
|
output = self.send("--offer", str(self.log), "71-siblings.sh", "46-neovim.sh")
|
|
self.assertIn("Your report's ID is CD-", output)
|
|
report_id = output.split("Your report's ID is ")[1].split(";")[0]
|
|
self.wait_for(lambda: self.gitea.attachments)
|
|
|
|
issue = self.gitea.issues[0]
|
|
self.assertEqual(issue["path"], "/api/v1/repos/Federated_Computer/coredesktop-reports/issues")
|
|
self.assertEqual(issue["auth"], "token secret")
|
|
self.assertTrue(issue["title"].startswith("Update failed in 71-siblings.sh, 46-neovim.sh"))
|
|
self.assertIn(report_id, issue["body"])
|
|
self.assertIn("Cloning into '~/x'...", issue["body"])
|
|
self.assertNotIn(str(self.home), issue["body"])
|
|
attachment = self.gitea.attachments[0]
|
|
self.assertEqual(attachment["path"], "/api/v1/repos/Federated_Computer/coredesktop-reports/issues/1/assets")
|
|
self.assertIn("Done, with errors in: 71-siblings.sh", attachment["text"])
|
|
self.assertFalse((self.tmp / "state" / report_id / "pending").exists())
|
|
|
|
def test_a_report_gitea_refuses_is_kept_and_filed_later(self):
|
|
self.gitea.fail = True
|
|
output = self.send(str(self.log))
|
|
self.assertIn("Your report's ID is CD-", output)
|
|
report_id = output.split("Your report's ID is ")[1].split(";")[0]
|
|
self.wait_for(lambda: not self.store.lock.locked() and (self.tmp / "state" / report_id / "pending").exists())
|
|
time.sleep(0.2)
|
|
self.assertEqual(self.gitea.issues, [])
|
|
|
|
self.gitea.fail = False
|
|
self.store.retry_and_prune()
|
|
self.assertEqual(len(self.gitea.issues), 1)
|
|
self.assertEqual(len(self.gitea.attachments), 1)
|
|
self.assertFalse((self.tmp / "state" / report_id / "pending").exists())
|
|
|
|
def test_an_attachment_that_failed_is_retried_without_a_second_issue(self):
|
|
report_id = self.store.save({"kind": "update", "version": "", "failed": "x", "kernel": "", "siblings": ""}, "log\n")
|
|
original = self.store.gitea.attach
|
|
self.store.gitea.attach = lambda *args: (_ for _ in ()).throw(OSError("Gitea is down"))
|
|
self.assertFalse(self.store.deliver(report_id))
|
|
self.store.gitea.attach = original
|
|
self.assertTrue(self.store.deliver(report_id))
|
|
self.assertEqual(len(self.gitea.issues), 1)
|
|
self.assertEqual(len(self.gitea.attachments), 1)
|
|
|
|
def post(self, body, content_type="multipart/form-data; boundary=x"):
|
|
request = urllib.request.Request(self.url, data=body, method="POST", headers={"Content-Type": content_type})
|
|
try:
|
|
with urllib.request.urlopen(request) as response:
|
|
return response.status, response.read().decode()
|
|
except urllib.error.HTTPError as error:
|
|
return error.code, error.read().decode()
|
|
|
|
def test_things_that_arent_reports_are_turned_away(self):
|
|
self.assertEqual(self.post(b"hello", "text/plain")[0], 400)
|
|
no_log = b'--x\r\nContent-Disposition: form-data; name="kind"\r\n\r\nupdate\r\n--x--\r\n'
|
|
self.assertEqual(self.post(no_log)[0], 400)
|
|
not_gzip = b'--x\r\nContent-Disposition: form-data; name="log"; filename="log.gz"\r\n\r\nplain\r\n--x--\r\n'
|
|
self.assertEqual(self.post(not_gzip), (400, "The log in this report couldn't be read.\n"))
|
|
self.assertEqual(self.gitea.issues, [])
|
|
|
|
def test_a_gzip_bomb_is_refused(self):
|
|
bomb = gzip.compress(b"\0" * (receiver.MAX_LOG_BYTES + 1))
|
|
body = (b'--x\r\nContent-Disposition: form-data; name="log"; filename="log.gz"\r\n'
|
|
b'Content-Type: application/gzip\r\n\r\n' + bomb + b'\r\n--x--\r\n')
|
|
self.assertEqual(self.post(body)[0], 413)
|
|
|
|
def test_too_many_reports_from_one_address_are_refused(self):
|
|
for _ in range(5):
|
|
self.assertIn("Your report's ID is", self.send(str(self.log)))
|
|
output = self.send(str(self.log))
|
|
self.assertIn("Couldn't send the report: Too many reports from this address", output)
|
|
|
|
def test_other_paths_are_not_found(self):
|
|
request = urllib.request.Request(self.url.replace("/report", "/other"), data=b"x", method="POST")
|
|
with self.assertRaises(urllib.error.HTTPError) as caught:
|
|
urllib.request.urlopen(request)
|
|
self.assertEqual(caught.exception.code, 404)
|
|
|
|
|
|
class PartsTest(unittest.TestCase):
|
|
def test_the_rate_limiter_counts_per_address_and_forgets_old_reports(self):
|
|
limiter = receiver.RateLimiter(per_address=[(3600, 2)], daily=3)
|
|
limiter.check("a", now=0)
|
|
limiter.check("a", now=1)
|
|
with self.assertRaises(receiver.Rejected):
|
|
limiter.check("a", now=2)
|
|
limiter.check("a", now=3700)
|
|
with self.assertRaises(receiver.Rejected): # the daily limit for everyone
|
|
limiter.check("b", now=3701)
|
|
|
|
def test_issue_text_cant_break_out_of_its_table_or_code_block(self):
|
|
details = {"kind": "update", "version": "v | 1", "failed": "a\nb", "kernel": "", "siblings": ""}
|
|
body = receiver.issue_body("CD-TEST23", details, "````\nlog\n", "now")
|
|
self.assertIn("| CoreDesktop | v \\| 1 |", body)
|
|
self.assertIn("| Failed | a<br>b |", body)
|
|
self.assertIn("`````\n````\nlog\n`````", body)
|
|
self.assertEqual(receiver.issue_title(details), "Update failed in a b (v | 1)")
|
|
|
|
def test_ids_are_easy_to_read_out(self):
|
|
for _ in range(100):
|
|
report_id = receiver.new_id()
|
|
self.assertRegex(report_id, r"^CD-[2-9A-HJ-NP-Z]{6}$")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|