davidandClaude Opus 5.5
43e227e3f3
shellcheck / shellcheck (push) Failing after 14s
The rate limit keeps addresses in memory, for at most a day; the log lines don't need them, so the privacy policy can say they aren't stored. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
362 lines
14 KiB
Python
362 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",
|
||
})
|
||
try:
|
||
with urllib.request.urlopen(request, timeout=self.timeout) as response:
|
||
return json.load(response)
|
||
except urllib.error.HTTPError as error:
|
||
# Gitea's reason, e.g. a token without the right permission, for the journal
|
||
with error:
|
||
detail = error.read(500).decode("utf-8", errors="replace").strip()
|
||
raise OSError(f"Gitea answered {error.code}: {detail or error.reason}") from None
|
||
|
||
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):
|
||
# Without the client's address: it's only kept, in memory, for the rate limit
|
||
print(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()
|