davidandClaude Opus 5.5
b7fb098de7
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>
355 lines
14 KiB
Python
355 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""Receive CoreDesktop error reports and file them as Gitea issues.
|
||
|
||
The `report` command on CoreDesktop machines POSTs reports here, through Caddy
|
||
on federated.sh. See the top of `report` for the request format. Each report
|
||
is saved under the state directory first, so none is lost if Gitea is down,
|
||
then filed as an issue in a private repository, with the full log attached.
|
||
Reports Gitea didn't take are tried again every few minutes.
|
||
|
||
Runs with only the standard library. Settings come from the environment:
|
||
|
||
GITEA_URL e.g. https://gitea.federated.computer
|
||
GITEA_REPO e.g. Federated_Computer/coredesktop-reports
|
||
GITEA_TOKEN a token that can create issues in that repository
|
||
LISTEN_PORT port on 127.0.0.1 to listen on (default 8650)
|
||
STATE_DIRECTORY where reports are kept (systemd sets this)
|
||
|
||
Saved reports are deleted after KEEP_DAYS days; their Gitea issues stay.
|
||
"""
|
||
|
||
import email.parser
|
||
import email.policy
|
||
import gzip
|
||
import http.server
|
||
import io
|
||
import json
|
||
import os
|
||
import re
|
||
import secrets
|
||
import shutil
|
||
import sys
|
||
import threading
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
from collections import deque
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
MAX_REQUEST_BYTES = 11 * 1024 * 1024 # the client sends at most 10 MB of log
|
||
MAX_LOG_BYTES = 64 * 1024 * 1024 # uncompressed, so a gzip bomb can't fill memory
|
||
FIELD_LIMITS = {"kind": 20, "version": 100, "failed": 1000, "kernel": 100, "siblings": 4000}
|
||
KINDS = {"install", "update", "other"}
|
||
PER_ADDRESS_LIMITS = [(3600, 5), (86400, 20)] # (seconds, reports) per client address
|
||
DAILY_LIMIT = 500 # from everyone together
|
||
EXCERPT_LINES = 150
|
||
RETRY_SECONDS = 600
|
||
KEEP_DAYS = 90
|
||
|
||
|
||
class Rejected(Exception):
|
||
"""A request we turn away, with the status and the message for the user."""
|
||
|
||
def __init__(self, status, message):
|
||
super().__init__(message)
|
||
self.status = status
|
||
self.message = message
|
||
|
||
|
||
class RateLimiter:
|
||
def __init__(self, per_address=PER_ADDRESS_LIMITS, daily=DAILY_LIMIT):
|
||
self.per_address = per_address
|
||
self.daily = daily
|
||
self.times = {}
|
||
self.everyone = deque()
|
||
self.lock = threading.Lock()
|
||
|
||
def check(self, address, now=None):
|
||
"""Count a report from address, or raise Rejected if it's one too many."""
|
||
now = time.time() if now is None else now
|
||
longest = max(window for window, _ in self.per_address)
|
||
with self.lock:
|
||
while self.everyone and self.everyone[0] <= now - 86400:
|
||
self.everyone.popleft()
|
||
if len(self.everyone) >= self.daily:
|
||
raise Rejected(429, "We're getting a lot of reports right now; please try again tomorrow.")
|
||
times = self.times.setdefault(address, deque())
|
||
while times and times[0] <= now - longest:
|
||
times.popleft()
|
||
for window, limit in self.per_address:
|
||
if sum(1 for t in times if t > now - window) >= limit:
|
||
raise Rejected(429, "Too many reports from this address; please try again later.")
|
||
times.append(now)
|
||
self.everyone.append(now)
|
||
# Forget addresses with nothing recent, so the table can't grow without end
|
||
if len(self.times) > 10000:
|
||
self.times = {a: t for a, t in self.times.items() if t and t[-1] > now - longest}
|
||
|
||
|
||
def parse_multipart(content_type, body):
|
||
"""The fields of a multipart/form-data body, as {name: bytes}."""
|
||
if not content_type.lower().startswith("multipart/form-data"):
|
||
raise Rejected(400, "That doesn't look like a CoreDesktop report.")
|
||
head = f"Content-Type: {content_type}\r\nMIME-Version: 1.0\r\n\r\n".encode()
|
||
message = email.parser.BytesParser(policy=email.policy.HTTP).parsebytes(head + body)
|
||
if not message.is_multipart():
|
||
raise Rejected(400, "That doesn't look like a CoreDesktop report.")
|
||
fields = {}
|
||
for part in message.iter_parts():
|
||
name = part.get_param("name", header="content-disposition")
|
||
if name and name not in fields:
|
||
fields[name] = part.get_payload(decode=True) or b""
|
||
return fields
|
||
|
||
|
||
def clean_text(value, limit):
|
||
"""value as text: no control characters but newlines and tabs, at most limit characters."""
|
||
text = value.decode("utf-8", errors="replace")
|
||
text = re.sub(r"[^\S\n\t ]|[\x00-\x08\x0b-\x1f\x7f]", "", text)
|
||
return text[:limit].strip()
|
||
|
||
|
||
def read_report(fields):
|
||
"""The report in fields: its details as text, and its log, uncompressed."""
|
||
if "log" not in fields:
|
||
raise Rejected(400, "That doesn't look like a CoreDesktop report.")
|
||
details = {name: clean_text(fields.get(name, b""), limit) for name, limit in FIELD_LIMITS.items()}
|
||
if details["kind"] not in KINDS:
|
||
details["kind"] = "other"
|
||
try:
|
||
with gzip.GzipFile(fileobj=io.BytesIO(fields["log"])) as stream:
|
||
log = stream.read(MAX_LOG_BYTES + 1)
|
||
except (OSError, EOFError):
|
||
raise Rejected(400, "The log in this report couldn't be read.")
|
||
if len(log) > MAX_LOG_BYTES:
|
||
raise Rejected(413, "This report's log is too big.")
|
||
return details, log.decode("utf-8", errors="replace")
|
||
|
||
|
||
def new_id():
|
||
"""A report ID such as CD-7Q2K9M: easy to read out, with no 0/O or 1/I."""
|
||
alphabet = "23456789ABCDEFGHJKLMNPQRSTUVWXYZ"
|
||
return "CD-" + "".join(secrets.choice(alphabet) for _ in range(6))
|
||
|
||
|
||
def fence(text):
|
||
"""A code fence longer than any run of backticks in text."""
|
||
longest = max((len(run) for run in re.findall(r"`+", text)), default=0)
|
||
return "`" * max(3, longest + 1)
|
||
|
||
|
||
def cell(text):
|
||
"""text for one cell of a Markdown table."""
|
||
return text.replace("\\", "\\\\").replace("|", "\\|").replace("\n", "<br>") or "–"
|
||
|
||
|
||
def issue_title(details):
|
||
failed = details["failed"].replace("\n", " ") or "an unknown step"
|
||
title = f"{details['kind'].capitalize()} failed in {failed}"
|
||
if details["version"]:
|
||
title += f" ({details['version']})"
|
||
return title[:200]
|
||
|
||
|
||
def issue_body(report_id, details, log, received):
|
||
lines = log.rstrip("\n").split("\n")
|
||
excerpt = "\n".join(lines[-EXCERPT_LINES:])
|
||
marks = fence(excerpt)
|
||
shown = "Its last %d lines:" % EXCERPT_LINES if len(lines) > EXCERPT_LINES else "The log:"
|
||
return "\n".join([
|
||
f"Report **{report_id}**, received {received}.",
|
||
"",
|
||
"| | |",
|
||
"|---|---|",
|
||
f"| Kind | {cell(details['kind'])} |",
|
||
f"| CoreDesktop | {cell(details['version'])} |",
|
||
f"| Failed | {cell(details['failed'])} |",
|
||
f"| Kernel | {cell(details['kernel'])} |",
|
||
f"| Siblings | {cell(details['siblings'])} |",
|
||
"",
|
||
f"The full log ({len(lines)} lines) is attached. {shown}",
|
||
"",
|
||
marks,
|
||
excerpt,
|
||
marks,
|
||
"",
|
||
"Sent with `report` from a CoreDesktop machine. Its user name, host name, home folder, "
|
||
"email and IP addresses were replaced before sending.",
|
||
])
|
||
|
||
|
||
class Gitea:
|
||
def __init__(self, url, repo, token, timeout=30):
|
||
self.base = f"{url.rstrip('/')}/api/v1/repos/{repo}"
|
||
self.token = token
|
||
self.timeout = timeout
|
||
|
||
def _request(self, path, data, content_type):
|
||
request = urllib.request.Request(self.base + path, data=data, method="POST", headers={
|
||
"Authorization": f"token {self.token}",
|
||
"Content-Type": content_type,
|
||
"Accept": "application/json",
|
||
})
|
||
with urllib.request.urlopen(request, timeout=self.timeout) as response:
|
||
return json.load(response)
|
||
|
||
def create_issue(self, title, body):
|
||
"""The new issue's number."""
|
||
data = json.dumps({"title": title, "body": body}).encode()
|
||
return self._request("/issues", data, "application/json")["number"]
|
||
|
||
def attach(self, number, filename, text):
|
||
boundary = "coredesktop-" + secrets.token_hex(16)
|
||
data = b"".join([
|
||
f"--{boundary}\r\n".encode(),
|
||
f'Content-Disposition: form-data; name="attachment"; filename="{filename}"\r\n'.encode(),
|
||
b"Content-Type: text/plain; charset=utf-8\r\n\r\n",
|
||
text.encode("utf-8"),
|
||
f"\r\n--{boundary}--\r\n".encode(),
|
||
])
|
||
self._request(f"/issues/{number}/assets", data, f"multipart/form-data; boundary={boundary}")
|
||
|
||
|
||
class Store:
|
||
"""Reports on disk: <id>/report.json and <id>/log.txt, and <id>/pending
|
||
until it's filed in Gitea."""
|
||
|
||
def __init__(self, directory, gitea):
|
||
self.directory = Path(directory)
|
||
self.gitea = gitea
|
||
self.lock = threading.Lock() # one delivery at a time, so none is filed twice
|
||
|
||
def save(self, details, log):
|
||
received = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
|
||
for _ in range(10):
|
||
report_id = new_id()
|
||
folder = self.directory / report_id
|
||
try:
|
||
folder.mkdir(parents=True)
|
||
break
|
||
except FileExistsError:
|
||
continue
|
||
else:
|
||
raise Rejected(503, "The report couldn't be saved; please try again later.")
|
||
(folder / "log.txt").write_text(log, encoding="utf-8")
|
||
(folder / "report.json").write_text(json.dumps({**details, "id": report_id, "received": received}))
|
||
(folder / "pending").touch()
|
||
return report_id
|
||
|
||
def deliver(self, report_id):
|
||
"""File one saved report in Gitea; True once it's filed."""
|
||
folder = self.directory / report_id
|
||
with self.lock:
|
||
if not (folder / "pending").exists():
|
||
return True
|
||
report = json.loads((folder / "report.json").read_text())
|
||
log = (folder / "log.txt").read_text(encoding="utf-8")
|
||
try:
|
||
number = report.get("issue")
|
||
if not number:
|
||
number = self.gitea.create_issue(issue_title(report), issue_body(report_id, report, log, report["received"]))
|
||
# Remember the issue, so a failed attachment doesn't make a second one
|
||
report["issue"] = number
|
||
(folder / "report.json").write_text(json.dumps(report))
|
||
self.gitea.attach(number, f"{report_id}-{report['kind']}.log", log)
|
||
except (urllib.error.URLError, OSError, ValueError, KeyError) as error:
|
||
print(f"{report_id}: couldn't file it in Gitea yet: {error}", file=sys.stderr, flush=True)
|
||
return False
|
||
(folder / "pending").unlink()
|
||
print(f"{report_id}: filed as issue #{number}", file=sys.stderr, flush=True)
|
||
return True
|
||
|
||
def retry_and_prune(self):
|
||
"""Try the reports Gitea hasn't taken again; delete filed ones older than KEEP_DAYS."""
|
||
if not self.directory.is_dir():
|
||
return
|
||
cutoff = time.time() - KEEP_DAYS * 86400
|
||
for folder in sorted(self.directory.iterdir()):
|
||
if not folder.is_dir():
|
||
continue
|
||
if (folder / "pending").exists():
|
||
self.deliver(folder.name)
|
||
elif folder.stat().st_mtime < cutoff:
|
||
shutil.rmtree(folder, ignore_errors=True)
|
||
|
||
|
||
def make_handler(store, limiter):
|
||
class Handler(http.server.BaseHTTPRequestHandler):
|
||
timeout = 60
|
||
server_version = "coredesktop-reports"
|
||
sys_version = ""
|
||
|
||
def reply(self, status, text):
|
||
data = (text + "\n").encode()
|
||
self.send_response(status)
|
||
self.send_header("Content-Type", "text/plain; charset=utf-8")
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.end_headers()
|
||
self.wfile.write(data)
|
||
|
||
def client_address_seen(self):
|
||
# Behind Caddy, the client is the first address it forwarded
|
||
forwarded = self.headers.get("X-Forwarded-For", "")
|
||
if self.client_address[0] in ("127.0.0.1", "::1") and forwarded:
|
||
return forwarded.split(",")[0].strip()
|
||
return self.client_address[0]
|
||
|
||
def do_GET(self):
|
||
if self.path == "/report/health":
|
||
self.reply(200, "ok")
|
||
else:
|
||
self.reply(404, "Not found.")
|
||
|
||
def do_POST(self):
|
||
if self.path != "/report":
|
||
self.reply(404, "Not found.")
|
||
return
|
||
try:
|
||
length = int(self.headers.get("Content-Length", ""))
|
||
except ValueError:
|
||
self.reply(411, "A report needs a Content-Length.")
|
||
return
|
||
if length > MAX_REQUEST_BYTES:
|
||
self.reply(413, "This report is too big.")
|
||
return
|
||
try:
|
||
limiter.check(self.client_address_seen())
|
||
fields = parse_multipart(self.headers.get("Content-Type", ""), self.rfile.read(length))
|
||
details, log = read_report(fields)
|
||
report_id = store.save(details, log)
|
||
except Rejected as rejected:
|
||
self.reply(rejected.status, rejected.message)
|
||
return
|
||
self.reply(200, report_id)
|
||
# Filed after replying, so the user isn't kept waiting on Gitea
|
||
threading.Thread(target=store.deliver, args=(report_id,), daemon=True).start()
|
||
|
||
def log_message(self, format, *args):
|
||
print(f"{self.client_address_seen()} {format % args}", file=sys.stderr, flush=True)
|
||
|
||
return Handler
|
||
|
||
|
||
def main():
|
||
missing = [name for name in ("GITEA_URL", "GITEA_REPO", "GITEA_TOKEN") if not os.environ.get(name)]
|
||
if missing:
|
||
sys.exit(f"Set {', '.join(missing)} (see the top of this file).")
|
||
gitea = Gitea(os.environ["GITEA_URL"], os.environ["GITEA_REPO"], os.environ["GITEA_TOKEN"])
|
||
store = Store(os.environ.get("STATE_DIRECTORY", "./reports"), gitea)
|
||
port = int(os.environ.get("LISTEN_PORT", "8650"))
|
||
server = http.server.ThreadingHTTPServer(("127.0.0.1", port), make_handler(store, RateLimiter()))
|
||
|
||
def retry_loop():
|
||
while True:
|
||
store.retry_and_prune()
|
||
time.sleep(RETRY_SECONDS)
|
||
|
||
threading.Thread(target=retry_loop, daemon=True).start()
|
||
print(f"Listening on 127.0.0.1:{port}, filing reports in {os.environ['GITEA_REPO']}", file=sys.stderr, flush=True)
|
||
server.serve_forever()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|